# 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):

```json
// .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):

```yaml
# .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

```python
# 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

```python
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

```python
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

```python
# 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

```python
# 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

```python
# 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

```python
# 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

```python
# 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

```python
# 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

```bash
# 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.