integration-strategy.md 23 KB
Integration Strategy: lambdagentpaas as a Plugin for Existing Frameworks
Date: 2026-04-04 Status: Proposal Core Principle: Don't sell a framework — sell a safety layer. Users keep LangChain/CrewAI/AutoGen, we add static analysis on top.
1. Positioning
lambdagentpaas should not compete with LangChain/CrewAI/AutoGen as an agent framework. Instead, it should serve as a static analysis layer — like TypeScript for JavaScript, mypy for Python, or ESLint for code quality.
| Analogy | Original Language | Checker Tool | Must users switch? |
|---|---|---|---|
| TypeScript | JavaScript | tsc | No |
| mypy | Python | mypy | No |
| ESLint | JavaScript | eslint | No |
| lambdagent-guard | LangChain/CrewAI/AutoGen | lint + type + cost | No |
Users continue writing agents in their preferred framework. lambdagentpaas extracts the configuration, compiles it to a λA term, and runs static analysis — type checking, cost prediction, loop termination verification, parallel safety — before a single LLM call is made.
2. Four Integration Modes
2.1 MCP Server (Highest Priority)
Expose static analysis as MCP tools. All AI IDEs that support MCP (Claude Code, Cursor, Windsurf, VS Code Copilot) gain lambdagent capabilities with one config line.
User setup (one line in settings):
// .claude/settings.json (Claude Code)
// .cursor/mcp.json (Cursor)
{
"mcpServers": {
"lambdagent-analyzer": {
"command": "uvx",
"args": ["lambdagent-mcp-server"]
}
}
}
Tools exposed via MCP:
lint_agent_config
Lint an agent YAML/JSON config for structural defects. Works with LangChain, CrewAI, AutoGen, Dify, and generic configs. Detects: missing terminate conditions, type mismatches, dead routes, empty loops, and 20+ other defect patterns.
Input: { "config_path": "agents/security_scanner.yml" }
Output: {
"framework": "crewai",
"errors": [
{
"rule": "L004a",
"level": "ERROR",
"message": "No terminate tool in ReAct loop (maxSteps=200)",
"lambda_meaning": "fix₂₀₀ has no base case λx.x",
"line": 15,
"fix": "Add 'terminate' to localTools list"
}
],
"warnings": [...],
"summary": "2 errors, 3 warnings"
}
estimate_agent_cost
Estimate worst-case cost of an agent pipeline BEFORE execution. Returns token upper bound, latency estimate, dollar cost, and end-to-end success probability.
Input: { "config_path": "agents/research_pipeline.yml" }
Output: {
"tokens_upper_bound": 164000,
"latency_upper_bound_sec": 173,
"cost_upper_bound_usd": 1.74,
"success_probability": 0.0016,
"breakdown": [
{ "stage": "parallel_scanners", "cost": 0.18, "probability": 0.021 },
{ "stage": "deep_analysis", "cost": 0.60, "probability": 0.12 },
{ "stage": "fix_and_review", "cost": 0.96, "probability": 0.62 }
],
"recommendation": "Success probability 0.16% is critically low. Primary bottleneck: 5 parallel scanners must ALL succeed (0.46^5=2.1%). Reduce to 3 scanners or lower maxSteps."
}
check_agent_types
Type-check an agent pipeline. Verifies that each stage's output type is compatible with the next stage's input type (Paper III T-Compose rule).
Input: { "config_path": "agents/data_pipeline.yml" }
Output: {
"type_safe": false,
"errors": [
{
"stage": 2,
"composition": "scanner >> analyzer",
"output_type": "Json(object({results: array(string)}))",
"input_type": "Str",
"error": "Json(object) is not subtype of Str",
"fix": "Add a Json-to-Str adapter between scanner and analyzer, or change analyzer to accept Json input"
}
]
}
check_parallel_safety
Check if parallel agents have store-independence (no shared mutable state). Prevents data corruption from race conditions (Paper II Proposition 30).
Input: { "config_path": "agents/multi_agent.yml" }
Output: {
"safe": false,
"conflicts": [
{
"agent_a": "researcher",
"agent_b": "code_analyzer",
"shared_keys": ["shared_doc"],
"risk": "Both agents write to 'shared_doc'. Last-write-wins race condition.",
"fix": "Use separate keys: 'research_notes' and 'code_analysis'"
}
]
}
Implementation: ~300 lines. All four tools call existing functions: lint_config(), compute_grade(), type_check(), check_store_independence().
2.2 GitHub Action (CI/CD Gate)
Agent config lint runs on every pull request, blocking merges with structural defects.
User setup (add one workflow file):
# .github/workflows/agent-lint.yml
name: Agent Config Lint
on:
pull_request:
paths:
- '**/*.yml'
- '**/*.yaml'
- '**/*.json'
jobs:
lint:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: lambdagent/agent-lint-action@v1
with:
# Which directories to scan
paths: |
agents/
configs/
# Fail the check on errors (options: error | warn | info)
fail-on: error
# Auto-detect framework (options: auto | crewai | langchain | autogen | dify)
frameworks: auto
# Fail if worst-case cost per run exceeds threshold
cost-threshold: 5.00
# Enable type checking across pipeline stages
type-check: true
# Enable parallel safety verification
parallel-check: true
PR comment output:
## 🔍 Agent Config Lint Results
### ❌ agents/security_scanner.yml (2 errors, 1 warning)
| Level | Rule | Message | λA Meaning |
|-------|------|---------|------------|
| ❌ ERROR | L004a | No terminate tool in ReAct loop (maxSteps=200) | fix₂₀₀ has no base case λx.x — forced truncation |
| ❌ ERROR | T-COMPOSE | Stage 2 output `Json(object)` ≠ Stage 3 input `Str` | Composition type mismatch: B ≮: B' |
| ⚠️ WARN | COST | Worst-case cost $12.40/run (threshold: $5.00) | Graded type: (0.016, 164000, 173s, $12.40) |
### ✅ agents/data_fetcher.yml (clean)
### Summary
- 1/2 configs have errors
- Estimated total cost: $12.40 + $0.85 = $13.25 per full run
- Recommendation: Fix security_scanner.yml before merge
Implementation: ~200 lines (Docker action wrapping lambdagent lint --format json).
2.3 Python Middleware (Runtime Guard)
Non-invasive wrapper around existing framework objects. Users add 2 lines of code; the rest of their codebase stays unchanged.
LangChain Guard
# pip install lambdagent-guard
from langchain.agents import AgentExecutor, create_react_agent
from lambdagent_guard import guard_langchain
# === Original LangChain code (unchanged) ===
llm = ChatAnthropic(model="claude-sonnet-4-20250514")
tools = [search_tool, read_tool, write_tool]
agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, max_iterations=20)
# === Add guard (2 lines) ===
guarded = guard_langchain(executor,
cost_budget=5.00, # Pause if cost exceeds $5
type_check=True, # Verify tool I/O type compatibility
loop_detection=True, # Detect repeated identical states
cost_alert=lambda c: print(f"⚠️ Cost so far: ${c:.2f}")
)
# === Use exactly as before ===
result = guarded.invoke({"input": "Analyze this repository"})
CrewAI Guard
from crewai import Agent, Task, Crew
from lambdagent_guard import guard_crewai
# === Original CrewAI code (unchanged) ===
researcher = Agent(role="Researcher", goal="...", tools=[search])
writer = Agent(role="Writer", goal="...", tools=[write])
crew = Crew(agents=[researcher, writer], tasks=[...])
# === Add guard (2 lines) ===
guarded_crew = guard_crewai(crew,
cost_budget=10.00, # Budget ceiling
parallel_safety=True, # Check agent store independence
terminate_check=True, # Verify termination conditions exist
)
result = guarded_crew.kickoff()
AutoGen Guard
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from lambdagent_guard import guard_autogen
# === Original AutoGen code (unchanged) ===
assistant = AssistantAgent("assistant", llm_config=llm_config)
proxy = UserProxyAgent("user_proxy", code_execution_config={...})
chat = GroupChat(agents=[assistant, proxy], max_round=20)
manager = GroupChatManager(groupchat=chat)
# === Add guard (2 lines) ===
guarded_manager = guard_autogen(manager,
cost_budget=5.00,
empty_message_detection=True, # Catch AutoGen #108 blank-message loops
terminate_robustness=True, # Don't rely on exact string matching
)
proxy.initiate_chat(guarded_manager, message="Write a sorting algorithm")
How Middleware Works Internally
# lambdagent_guard/core.py
class GuardedExecutor:
"""Wraps any agent executor with lambdagent static/dynamic guards."""
def __init__(self, executor, config, **opts):
self.executor = executor
self.opts = opts
# Phase 1: Compile-time checks (before any execution)
term = from_config(config)
if opts.get("type_check"):
errors = type_check(term)
if errors:
raise AgentTypeError(
f"Pipeline type mismatch: {errors[0].description}\n"
f"This would crash at runtime after spending "
f"${errors[0].wasted_cost:.2f} on preceding stages."
)
grade = compute_grade(term)
if grade.cost > opts.get("cost_budget", float("inf")):
raise CostBudgetExceeded(
f"Worst-case cost ${grade.cost:.2f} exceeds "
f"budget ${opts['cost_budget']:.2f}\n"
f"Breakdown: {grade.per_stage_summary()}"
)
if opts.get("parallel_safety"):
conflicts = check_store_independence(term)
if conflicts:
raise StoreConflictError(conflicts)
# Phase 2: Runtime hooks (during execution)
self.cost_accumulator = CostVector(0, 0, 0)
self.state_history = []
def on_step(self, step_info):
"""Called before each LLM/tool invocation (CEK Yield equivalent)."""
# Cost monitoring
self.cost_accumulator += estimate_step_cost(step_info)
if self.opts.get("cost_alert"):
self.opts["cost_alert"](self.cost_accumulator.cost)
if self.cost_accumulator.cost > self.opts.get("cost_budget", float("inf")):
raise CostBudgetExceeded(self.cost_accumulator)
# Loop detection
if self.opts.get("loop_detection"):
state_hash = hash_state(step_info)
recent = self.state_history[-5:]
if recent.count(state_hash) >= 3:
raise InfiniteLoopDetected(
f"Identical state detected {recent.count(state_hash)} "
f"times in last 5 steps. Agent is not making progress."
)
self.state_history.append(state_hash)
Implementation: ~800 lines per framework (extractor + guard wrapper).
2.4 VS Code / JetBrains Extension
Real-time lint on YAML file save. Shows inline diagnostics in the editor.
User experience:
# User edits agents/security_scanner.yml in VS Code
agents/security_scanner.yml
│
├─ Line 5: ❌ L004a: No terminate tool — ReAct loop may never exit
│ λA: fix₂₀₀ has no base case λx.x
│ Quick Fix: Add 'terminate' to localTools
│
├─ Line 12: ❌ T-COMPOSE: output(scanner)=Json ≠ input(analyzer)=Str
│ Quick Fix: Add json_to_str adapter between stages
│
├─ Line 18: ⚠️ COST: maxSteps=200 × claude-opus = $6.00/run upper bound
│ Quick Fix: Change maxSteps to 20
│
└─ Line 23: ⚠️ L013: Route missing default branch — non-exhaustive dispatch
Quick Fix: Add 'default' route
Status bar: λA: 2 errors, 2 warnings | Est. cost: $6.00/run | Success: 0.16%
Implementation: ~500 lines (VS Code extension calling lambdagent lint --format json via Language Server Protocol).
3. Framework Config Extractors
The key technical component that enables all integration modes: extracting λA-compatible configurations from each framework's runtime objects.
Architecture
lambdagent/extractors/
├── base.py # Abstract extractor interface
├── langchain_extractor.py # LangChain Agent → normalized config
├── crewai_extractor.py # CrewAI Crew → normalized config
├── autogen_extractor.py # AutoGen GroupChat → normalized config
└── dify_extractor.py # Dify workflow → normalized config
Base Interface
# lambdagent/extractors/base.py
from abc import ABC, abstractmethod
from typing import Dict, Any
class FrameworkExtractor(ABC):
"""Extract lambdagent-compatible config from a framework object."""
@abstractmethod
def extract(self, framework_object: Any) -> Dict[str, Any]:
"""Convert framework-specific agent to normalized YAML config dict.
The returned dict must conform to lambdagent YAML schema:
- type: simple | react | chain | router | parallel
- model: { name, temperature, maxTokens }
- systemPrompt: str
- react: { maxSteps }
- mcp: { localTools: [...] }
- memory: { enabled, strategy, size }
"""
...
@abstractmethod
def detect(self, obj: Any) -> bool:
"""Return True if obj is an instance of this framework's agent."""
...
LangChain Extractor
# lambdagent/extractors/langchain_extractor.py
class LangChainExtractor(FrameworkExtractor):
def detect(self, obj):
return hasattr(obj, 'agent') and hasattr(obj, 'tools') and \
hasattr(obj, 'max_iterations')
def extract(self, executor) -> dict:
# Extract model info
llm = executor.agent.llm_chain.llm if hasattr(executor.agent, 'llm_chain') \
else executor.agent.llm
model_name = getattr(llm, 'model_name', getattr(llm, 'model', 'unknown'))
temperature = getattr(llm, 'temperature', 0.0)
# Extract tools
tool_names = [t.name for t in executor.tools]
has_terminate = any(
name in ('terminate', 'final_answer', 'human')
for name in tool_names
)
# Extract prompt
prompt_template = ""
if hasattr(executor.agent, 'llm_chain') and \
hasattr(executor.agent.llm_chain, 'prompt'):
prompt_template = executor.agent.llm_chain.prompt.template
return {
"type": "react",
"model": {
"name": model_name,
"temperature": temperature,
},
"systemPrompt": prompt_template,
"react": {
"maxSteps": executor.max_iterations or 15,
},
"mcp": {
"localTools": tool_names + (["terminate"] if has_terminate else []),
},
}
CrewAI Extractor
# lambdagent/extractors/crewai_extractor.py
class CrewAIExtractor(FrameworkExtractor):
def detect(self, obj):
return hasattr(obj, 'agents') and hasattr(obj, 'tasks') and \
hasattr(obj, 'kickoff')
def extract(self, crew) -> dict:
agents = []
for agent in crew.agents:
agent_config = {
"type": "react",
"model": {
"name": getattr(agent, 'llm', {}).get('model', 'unknown') \
if isinstance(getattr(agent, 'llm', None), dict) \
else str(getattr(agent, 'llm', 'unknown')),
},
"systemPrompt": f"Role: {agent.role}\nGoal: {agent.goal}\n"
f"Backstory: {agent.backstory}",
"react": {
"maxSteps": getattr(agent, 'max_iter', 25),
},
"mcp": {
"localTools": [t.name for t in (agent.tools or [])],
},
}
agents.append(agent_config)
# Determine crew execution pattern
process = getattr(crew, 'process', 'sequential')
if process == 'sequential':
return {
"type": "chain",
"steps": agents,
}
elif process == 'hierarchical':
return {
"type": "router",
"classifier": agents[0], # manager agent
"routes": {a["systemPrompt"][:20]: a for a in agents[1:]},
}
else:
return {
"type": "parallel",
"agents": agents,
}
AutoGen Extractor
# lambdagent/extractors/autogen_extractor.py
class AutoGenExtractor(FrameworkExtractor):
def detect(self, obj):
return hasattr(obj, 'groupchat') or \
(hasattr(obj, 'llm_config') and hasattr(obj, 'system_message'))
def extract(self, manager) -> dict:
chat = manager.groupchat
agents = []
for agent in chat.agents:
agent_config = {
"type": "react",
"model": {
"name": self._extract_model(agent),
},
"systemPrompt": getattr(agent, 'system_message', ''),
"react": {
"maxSteps": getattr(agent, 'max_consecutive_auto_reply', 10),
},
"mcp": {
"localTools": self._extract_tools(agent),
},
}
agents.append(agent_config)
# GroupChat is a multi-agent loop
termination_msg = getattr(chat, 'is_termination_msg', None)
return {
"type": "parallel", # GroupChat agents interact
"agents": agents,
"multiagent": {
"maxRounds": getattr(chat, 'max_round', 10),
"terminationCondition": "is_termination_msg" if termination_msg else None,
},
}
def _extract_model(self, agent):
llm_config = getattr(agent, 'llm_config', {})
if isinstance(llm_config, dict):
config_list = llm_config.get('config_list', [{}])
if config_list:
return config_list[0].get('model', 'unknown')
return 'unknown'
def _extract_tools(self, agent):
funcs = getattr(agent, '_function_map', {})
return list(funcs.keys()) if funcs else []
Auto-Detection
# lambdagent/extractors/__init__.py
from .langchain_extractor import LangChainExtractor
from .crewai_extractor import CrewAIExtractor
from .autogen_extractor import AutoGenExtractor
_EXTRACTORS = [
LangChainExtractor(),
CrewAIExtractor(),
AutoGenExtractor(),
]
def extract_config(framework_object) -> dict:
"""Auto-detect framework and extract normalized config."""
for extractor in _EXTRACTORS:
if extractor.detect(framework_object):
return extractor.extract(framework_object)
raise UnsupportedFrameworkError(
f"Cannot extract config from {type(framework_object).__name__}. "
f"Supported: LangChain AgentExecutor, CrewAI Crew, AutoGen GroupChatManager."
)
4. REST API Endpoints (New)
Add dedicated lint/analysis endpoints to agentpaas API:
POST /api/v1/analyze/lint
Body: { "config": <YAML string or dict>, "framework": "auto" }
Returns: { "framework": "crewai", "errors": [...], "warnings": [...] }
POST /api/v1/analyze/type-check
Body: { "config": <YAML string or dict> }
Returns: { "type_safe": true/false, "errors": [...] }
POST /api/v1/analyze/cost
Body: { "config": <YAML string or dict> }
Returns: { "grade": { "p": 0.016, "t": 164000, "l": 173, "m": 1.74 }, "breakdown": [...] }
POST /api/v1/analyze/parallel-safety
Body: { "config": <YAML string or dict> }
Returns: { "safe": true/false, "conflicts": [...] }
POST /api/v1/analyze/full
Body: { "config": <YAML string or dict>, "framework": "auto" }
Returns: { "lint": {...}, "types": {...}, "cost": {...}, "parallel": {...} }
These endpoints power the GitHub Action (calls /api/v1/analyze/full) and can be used by any HTTP client.
5. Implementation Priority and Effort
| Mode | Effort | User Reach | Dependencies |
|---|---|---|---|
| MCP Server | ~300 LOC, 1-2 days | Claude Code, Cursor, Windsurf, all MCP clients | lint_config(), compute_grade(), type_check() (all exist) |
| GitHub Action | ~200 LOC, 1 day | All GitHub projects | CLI lambdagent lint (exists) |
| REST API endpoints | ~200 LOC, 1 day | Any HTTP client | Existing FastAPI app |
| Python middleware | ~800 LOC/framework, 1 week | LangChain/CrewAI/AutoGen users | Framework extractors (new) |
| VS Code extension | ~500 LOC, 2-3 days | VS Code users | CLI (exists) |
Recommended order: MCP Server → GitHub Action → REST endpoints → Python middleware → VS Code extension.
6. Distribution
PyPI Packages
lambdagent # Core DSL + compiler + lint + type checker
lambdagent-mcp-server # MCP Server (standalone, no framework deps)
lambdagent-guard # Python middleware for LangChain/CrewAI/AutoGen
Other Channels
GitHub Marketplace # lambdagent/agent-lint-action (GitHub Action)
VS Code Marketplace # lambdagent.agent-lint (VS Code extension)
Docker Hub # lambdagent/analyzer (for CI/CD)
Installation Commands
# MCP Server (for AI IDEs)
uvx lambdagent-mcp-server
# CLI tool (for terminal users)
pip install lambdagent
lambdagent lint agents/
# Python middleware (for framework users)
pip install lambdagent-guard
# Then: from lambdagent_guard import guard_langchain
# GitHub Action (for CI/CD)
# Add .github/workflows/agent-lint.yml (see Section 2.2)
# Docker (for any environment)
docker run lambdagent/analyzer lint /configs/
7. What Real Bugs Each Mode Catches
Evidence from documented GitHub issues (see REAL_WORLD_DEFECTS.md):
| Integration Mode | Example Bug Caught | Real Issue | Savings |
|---|---|---|---|
| MCP Server | Developer asks "check my agent config" → type mismatch found | LangChain #10997 | Prevents runtime crash |
| GitHub Action | PR adds agent with maxSteps=200 → cost warning blocks merge | Claude Code #38029 | Prevents $342 session |
| Python middleware | guard detects 3 identical states → kills loop | AutoGen #108 | Prevents token burn |
| VS Code extension | Inline warning on save: "no terminate tool" | CrewAI #737 | Prevents infinite loop |
| REST API | CI pipeline calls /analyze/full on every deploy | Claude Code #34629 | Catches 10-20x cost anomaly |
8. Key Insight
The theoretical work (three papers, λA calculus, type system, CEK machine) is the engine. The integration modes are the steering wheel. Without integration, the engine sits in a lab. With integration, every LangChain/CrewAI/AutoGen developer gets a safety net — without changing a single line of their agent code.
The MCP Server alone, at ~300 lines of code, puts the full power of the λA type system, 26 lint rules, graded cost prediction, and parallel safety verification into every AI IDE on the market. That is the highest-leverage implementation task in the entire project.