adapters.py

"""Custom adapter module for supporting async context manager LangGraph adapters"""

import textwrap
import logging
from typing import Callable, Type, AsyncContextManager
from pydantic import BaseModel

# Setup logging
logger = logging.getLogger(__name__)

def create_langgraph_async_adapter(
    agent_instance: AsyncContextManager,
    name: str,
    description: str,
    input_schema: Type[BaseModel],
) -> Callable:
    """
    Create a LangGraph adapter that supports async context managers
    
    Args:
        agent_instance: Async context manager, like make_graph
        name: Tool name
        description: Tool description
        input_schema: Pydantic model for input data
        
    Returns:
        Adapted async function that can be called by MCP server
    """
    
    # Get input schema fields
    schema_fields = input_schema.model_fields

    # Create parameter string
    params_str = ", ".join(
        f"{field_name}: {getattr(field_info.annotation, '__name__', 'Any')}"
        for field_name, field_info in schema_fields.items()
    )

    # Create function body that directly returns async function
    body_str = textwrap.dedent(f"""
    async def run_agent({params_str}):
        inputs = input_schema({', '.join(f'{name}={name}' for name in schema_fields)})
        logger.info(f"Received request with projectRoot: {{inputs.projectRoot}}")
        logger.info(f"File parameter: {{inputs.file if hasattr(inputs, 'file') else None}}")
        
        async with agent_instance(inputs.projectRoot) as agent:
            logger.info(f"Invoking agent with prompt: {{inputs.prompt[:50]}}...")
            result = await agent.ainvoke({{"messages": [{{"role": "user", "content": inputs.prompt}}]}})
            logger.info("Agent invocation completed")
        return result
    """)

    # Create namespace
    namespace = {
        "input_schema": input_schema,
        "agent_instance": agent_instance,
        "logger": logger
    }

    # Execute function definition
    exec(body_str, namespace)

    # Get created function
    run_agent = namespace["run_agent"]

    # Add function metadata
    run_agent.__name__ = name
    run_agent.__doc__ = description

    return run_agent