LangChain MCP Adapters: Complete Integration Guide for AI Developers

# Install core LangChain package
pip install langchain

# Install LangChain community package with MCP support
pip install langchain-community

# Install MCP SDK for Python
pip install mcp

# Install additional dependencies for specific MCP servers
pip install httpx websockets aiohttp

# Optional: Install specific integrations
pip install langchain-openai langchain-anthropic

# API Keys for AI Models
OPENAI_API_KEY=your_openai_key_here
ANTHROPIC_API_KEY=your_anthropic_key_here

# MCP Server Configuration
MCP_SERVER_URL=http://localhost:3000
MCP_AUTH_TOKEN=your_mcp_auth_token

# Optional: Logging and Debug Settings
LOG_LEVEL=INFO
DEBUG_MODE=false

# Using npm
npm install langchain @langchain/community @modelcontextprotocol/sdk

# Or using yarn
yarn add langchain @langchain/community @modelcontextprotocol/sdk

# Install TypeScript definitions if using TypeScript
npm install --save-dev @types/node

# Optional: Install specific model integrations
npm install @langchain/openai @langchain/anthropic

from langchain.agents import initialize_agent, AgentType
from langchain.chat_models import ChatOpenAI
from langchain_community.tools.mcp import MCPTool
from mcp import ClientSession, StdioServerParameters
import asyncio

class MCPFileSystemAdapter:
    def __init__(self, server_path):
        self.server_path = server_path
        self.session = None
        
    async def connect(self):
        """Establish connection to MCP server"""
        server_params = StdioServerParameters(
            command="node",
            args=[self.server_path],
            env=None
        )
        
        self.session = ClientSession(server_params)
        await self.session.initialize()
        
        # List available tools from the MCP server
        tools_list = await self.session.list_tools()
        print(f"Connected to MCP server with {len(tools_list)} tools")
        return tools_list
    
    async def create_langchain_tools(self):
        """Convert MCP tools to LangChain tool format"""
        tools_list = await self.connect()
        langchain_tools = []
        
        for tool in tools_list:
            mcp_tool = MCPTool(
                name=tool.name,
                description=tool.description,
                session=self.session,
                tool_name=tool.name
            )
            langchain_tools.append(mcp_tool)
        
        return langchain_tools
    
    async def cleanup(self):
        """Close MCP connection"""
        if self.session:
            await self.session.close()

async def main():
    # Initialize MCP adapter
    mcp_adapter = MCPFileSystemAdapter("./mcp-server-filesystem")
    
    # Get LangChain-compatible tools
    tools = await mcp_adapter.create_langchain_tools()
    
    # Initialize AI model
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    # Create agent with MCP tools
    agent = initialize_agent(
        tools=tools,
        llm=llm,
        agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
        verbose=True
    )
    
    # Execute agent task
    result = await agent.arun(
        "Read the contents of example.txt and create a summary file"
    )
    
    print(f"Agent result: {result}")
    
    # Cleanup
    await mcp_adapter.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

import { ChatOpenAI } from "@langchain/openai";
import { initializeAgentExecutorWithOptions } from "langchain/agents";
import { DynamicStructuredTool } from "@langchain/core/tools";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
import { z } from "zod";

class MCPLangChainAdapter {
    private client: Client;
    private transport: StdioClientTransport;
    
    constructor(private serverCommand: string, private serverArgs: string[]) {
        this.client = new Client({
            name: "langchain-mcp-client",
            version: "1.0.0"
        }, {
            capabilities: {
                tools: {}
            }
        });
    }
    
    async connect(): Promise {
        this.transport = new StdioClientTransport({
            command: this.serverCommand,
            args: this.serverArgs
        });
        
        await this.client.connect(this.transport);
        console.log("Connected to MCP server successfully");
    }
    
    async getLangChainTools(): Promise {
        const toolsResponse = await this.client.listTools();
        const tools: DynamicStructuredTool[] = [];
        
        for (const tool of toolsResponse.tools) {
            const dynamicTool = new DynamicStructuredTool({
                name: tool.name,
                description: tool.description || "MCP Tool",
                schema: this.convertMCPSchemaToZod(tool.inputSchema),
                func: async (input: any) => {
                    const result = await this.client.callTool({
                        name: tool.name,
                        arguments: input
                    });
                    return JSON.stringify(result.content);
                }
            });
            
            tools.push(dynamicTool);
        }
        
        return tools;
    }
    
    private convertMCPSchemaToZod(schema: any): z.ZodType {
        // Convert JSON Schema to Zod schema
        if (schema.type === "object") {
            const shape: any = {};
            for (const [key, value] of Object.entries(schema.properties || {})) {
                shape[key] = z.string().describe((value as any).description || "");
            }
            return z.object(shape);
        }
        return z.object({});
    }
    
    async disconnect(): Promise {
        await this.client.close();
    }
}

async function runMCPAgent() {
    // Initialize MCP adapter
    const mcpAdapter = new MCPLangChainAdapter("node", [
        "./mcp-server-filesystem/index.js"
    ]);
    
    await mcpAdapter.connect();
    
    // Get tools from MCP server
    const tools = await mcpAdapter.getLangChainTools();
    
    // Initialize language model
    const model = new ChatOpenAI({
        modelName: "gpt-4",
        temperature: 0
    });
    
    // Create agent executor
    const executor = await initializeAgentExecutorWithOptions(
        tools,
        model,
        {
            agentType: "structured-chat-zero-shot-react-description",
            verbose: true
        }
    );
    
    // Execute agent task
    const result = await executor.invoke({
        input: "List all files in the current directory and count them"
    });
    
    console.log("Agent Result:", result.output);
    
    // Cleanup
    await mcpAdapter.disconnect();
}

runMCPAgent().catch(console.error);

import asyncio
from typing import Dict, List
from contextlib import asynccontextmanager
from mcp import ClientSession

class MCPConnectionPool:
    def __init__(self, max_connections: int = 10):
        self.max_connections = max_connections
        self.available_connections: List[ClientSession] = []
        self.in_use_connections: Dict[str, ClientSession] = {}
        self._lock = asyncio.Lock()
        
    async def get_connection(self, server_config) -> ClientSession:
        """Get an available connection from the pool"""
        async with self._lock:
            if self.available_connections:
                session = self.available_connections.pop()
                connection_id = id(session)
                self.in_use_connections[connection_id] = session
                return session
            
            if len(self.in_use_connections) < self.max_connections:
                session = ClientSession(server_config)
                await session.initialize()
                connection_id = id(session)
                self.in_use_connections[connection_id] = session
                return session
            
            # Wait for a connection to become available
            await asyncio.sleep(0.1)
            return await self.get_connection(server_config)
    
    async def return_connection(self, session: ClientSession):
        """Return a connection to the pool"""
        async with self._lock:
            connection_id = id(session)
            if connection_id in self.in_use_connections:
                del self.in_use_connections[connection_id]
                self.available_connections.append(session)
    
    @asynccontextmanager
    async def connection(self, server_config):
        """Context manager for automatic connection management"""
        session = await self.get_connection(server_config)
        try:
            yield session
        finally:
            await self.return_connection(session)
    
    async def close_all(self):
        """Close all connections in the pool"""
        all_connections = (
            self.available_connections + 
            list(self.in_use_connections.values())
        )
        for session in all_connections:
            await session.close()

# Usage example
pool = MCPConnectionPool(max_connections=5)

async def process_request(request_data):
    async with pool.connection(server_config) as session:
        # Use the session for MCP operations
        result = await session.call_tool("process_data", request_data)
        return result

import time
from typing import Any, Callable
from functools import wraps
import logging

logger = logging.getLogger(__name__)

class MCPToolWrapper:
    def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    def with_retry(self, func: Callable) -> Callable:
        """Decorator that adds retry logic to MCP tool calls"""
        @wraps(func)
        async def wrapper(*args, **kwargs) -> Any:
            last_exception = None
            
            for attempt in range(self.max_retries):
                try:
                    result = await func(*args, **kwargs)
                    return result
                    
                except ConnectionError as e:
                    last_exception = e
                    delay = self.base_delay * (2 ** attempt)
                    logger.warning(
                        f"Connection error on attempt {attempt + 1}: {e}. "
                        f"Retrying in {delay}s..."
                    )
                    await asyncio.sleep(delay)
                    
                except TimeoutError as e:
                    last_exception = e
                    logger.warning(
                        f"Timeout on attempt {attempt + 1}: {e}. "
                        f"Retrying..."
                    )
                    
                except Exception as e:
                    logger.error(f"Unexpected error in MCP tool call: {e}")
                    raise
            
            # All retries exhausted
            logger.error(
                f"Failed after {self.max_retries} attempts: {last_exception}"
            )
            raise last_exception
        
        return wrapper
    
    @staticmethod
    def validate_tool_response(response: dict) -> dict:
        """Validate and sanitize MCP tool responses"""
        if not isinstance(response, dict):
            raise ValueError("MCP tool must return a dictionary")
        
        if "error" in response:
            raise RuntimeError(f"Tool returned error: {response['error']}")
        
        if "content" not in response:
            raise ValueError("Tool response missing 'content' field")
        
        return response

# Usage with LangChain
class ResilientMCPTool(MCPTool):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.wrapper = MCPToolWrapper()
    
    @property
    def _run(self):
        original_run = super()._run
        return self.wrapper.with_retry(original_run)

from typing import Optional, Any
import hashlib
import json
from datetime import datetime, timedelta
from collections import OrderedDict

class MCPResponseCache:
    def __init__(self, max_size: int = 1000, default_ttl: int = 3600):
        self.cache = OrderedDict()
        self.max_size = max_size
        self.default_ttl = default_ttl
        self.stats = {"hits": 0, "misses": 0}
    
    def _generate_key(self, tool_name: str, arguments: dict) -> str:
        """Generate a unique cache key for tool calls"""
        key_data = {
            "tool": tool_name,
            "args": arguments
        }
        key_string = json.dumps(key_data, sort_keys=True)
        return hashlib.sha256(key_string.encode()).hexdigest()
    
    def get(self, tool_name: str, arguments: dict) -> Optional[Any]:
        """Retrieve cached response if available and not expired"""
        key = self._generate_key(tool_name, arguments)
        
        if key in self.cache:
            entry = self.cache[key]
            if datetime.now() < entry["expires_at"]:
                self.stats["hits"] += 1
                # Move to end (LRU)
                self.cache.move_to_end(key)
                return entry["data"]
            else:
                # Expired entry
                del self.cache[key]
        
        self.stats["misses"] += 1
        return None
    
    def set(
        self, 
        tool_name: str, 
        arguments: dict, 
        data: Any,
        ttl: Optional[int] = None
    ):
        """Store response in cache"""
        key = self._generate_key(tool_name, arguments)
        
        # Evict oldest entry if cache is full
        if len(self.cache) >= self.max_size and key not in self.cache:
            self.cache.popitem(last=False)
        
        expires_at = datetime.now() + timedelta(
            seconds=ttl or self.default_ttl
        )
        
        self.cache[key] = {
            "data": data,
            "expires_at": expires_at,
            "created_at": datetime.now()
        }
    
    def invalidate(self, tool_name: str, arguments: dict):
        """Invalidate specific cache entry"""
        key = self._generate_key(tool_name, arguments)
        if key in self.cache:
            del self.cache[key]
    
    def clear(self):
        """Clear all cache entries"""
        self.cache.clear()
        self.stats = {"hits": 0, "misses": 0}
    
    def get_stats(self) -> dict:
        """Get cache statistics"""
        total = self.stats["hits"] + self.stats["misses"]
        hit_rate = (
            self.stats["hits"] / total * 100 if total > 0 else 0
        )
        
        return {
            "hits": self.stats["hits"],
            "misses": self.stats["misses"],
            "hit_rate": f"{hit_rate:.2f}%",
            "size": len(self.cache)
        }

# Integration with MCP adapter
class CachedMCPAdapter:
    def __init__(self, session: ClientSession, cache: MCPResponseCache):
        self.session = session
        self.cache = cache
    
    async def call_tool(
        self, 
        tool_name: str, 
        arguments: dict,
        use_cache: bool = True,
        ttl: Optional[int] = None
    ) -> Any:
        """Call MCP tool with caching support"""
        if use_cache:
            cached = self.cache.get(tool_name, arguments)
            if cached is not None:
                return cached
        
        result = await self.session.call_tool(tool_name, arguments)
        
        if use_cache:
            self.cache.set(tool_name, arguments, result, ttl)
        
        return result

from langchain.agents import AgentExecutor, create_structured_chat_agent
from langchain.prompts import ChatPromptTemplate
from langchain_community.tools.mcp import MCPTool

class DataAnalysisAgent:
    def __init__(self, mcp_session, llm):
        self.mcp_session = mcp_session
        self.llm = llm
        self.tools = []
        
    async def initialize(self):
        """Set up database tools from MCP server"""
        tools_list = await self.mcp_session.list_tools()
        
        for tool in tools_list:
            if tool.name.startswith("database_"):
                mcp_tool = MCPTool(
                    name=tool.name,
                    description=tool.description,
                    session=self.mcp_session,
                    tool_name=tool.name
                )
                self.tools.append(mcp_tool)
        
        # Create specialized prompt for data analysis
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert data analyst with access to 
            database tools. When analyzing data:
            1. Break complex queries into smaller steps
            2. Verify data quality before analysis
            3. Present findings with visualizations when possible
            4. Explain your reasoning clearly
            
            Available tools: {tool_names}
            Tool descriptions: {tools}"""),
            ("human", "{input}"),
            ("assistant", "{agent_scratchpad}")
        ])
        
        self.agent = create_structured_chat_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=self.prompt
        )
        
        self.executor = AgentExecutor(
            agent=self.agent,
            tools=self.tools,
            verbose=True,
            max_iterations=10
        )
    
    async def analyze(self, query: str) -> dict:
        """Execute analysis based on natural language query"""
        result = await self.executor.ainvoke({
            "input": query,
            "tool_names": ", ".join([t.name for t in self.tools]),
            "tools": "\n".join([
                f"{t.name}: {t.description}" for t in self.tools
            ])
        })
        
        return {
            "query": query,
            "result": result["output"],
            "steps": result.get("intermediate_steps", [])
        }

# Usage example
async def run_analysis():
    agent = DataAnalysisAgent(mcp_session, llm)
    await agent.initialize()
    
    result = await agent.analyze(
        "Show me the top 5 products by revenue in Q4 2024, "
        "and compare with Q4 2023"
    )
    
    print(result["result"])

from pydantic import BaseModel, validator, Field
from typing import Any, Dict
import re

class SecureMCPToolInput(BaseModel):
    """Base class for validated MCP tool inputs"""
    
    @validator('*', pre=True)
    def sanitize_strings(cls, v):
        """Remove potentially dangerous characters from inputs"""
        if isinstance(v, str):
            # Remove control characters
            v = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', v)
            # Limit string length
            v = v[:10000]
        return v

class DatabaseQueryInput(SecureMCPToolInput):
    """Validated input for database operations"""
    table_name: str = Field(
        ..., 
        regex=r'^[a-zA-Z_][a-zA-Z0-9_]*$',
        description="Valid SQL table name"
    )
    conditions: Dict[str, Any] = Field(default_factory=dict)
    limit: int = Field(default=100, ge=1, le=1000)
    
    @validator('conditions')
    def validate_conditions(cls, v):
        """Ensure conditions don't contain SQL injection"""
        for key, value in v.items():
            if isinstance(value, str):
                dangerous_patterns = [
                    r';.*--',
                    r'union.*select',
                    r'drop.*table',
                    r'exec.*\('
                ]
                for pattern in dangerous_patterns:
                    if re.search(pattern, value, re.IGNORECASE):
                        raise ValueError(
                            f"Potentially dangerous pattern in: {key}"
                        )
        return v

class FileOperationInput(SecureMCPToolInput):
    """Validated input for file operations"""
    file_path: str = Field(..., description="File path to operate on")
    
    @validator('file_path')
    def validate_path(cls, v):
        """Prevent path traversal attacks"""
        if '..' in v or v.startswith('/'):
            raise ValueError("Invalid file path")
        
        # Whitelist allowed directories
        allowed_dirs = ['uploads', 'data', 'temp']
        if not any(v.startswith(d) for d in allowed_dirs):
            raise ValueError(
                f"Path must start with: {', '.join(allowed_dirs)}"
            )
        
        return v

# Integration with MCP tools
class SecureMCPTool:
    def __init__(self, tool_name: str, input_model: type[BaseModel]):
        self.tool_name = tool_name
        self.input_model = input_model
    
    async def execute(self, raw_input: dict) -> Any:
        """Execute tool with validated input"""
        try:
            # Validate input
            validated = self.input_model(**raw_input)
            
            # Execute with validated data
            result = await self.session.call_tool(
                self.tool_name,
                validated.dict()
            )
            
            return result
            
        except ValueError as e:
            raise SecurityError(f"Input validation failed: {e}")

import asyncio
from typing import List, Any, Dict

class ParallelMCPExecutor:
    def __init__(self, session, max_concurrent: int = 10):
        self.session = session
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def execute_tool(
        self, 
        tool_name: str, 
        arguments: dict
    ) -> Dict[str, Any]:
        """Execute single tool with rate limiting"""
        async with self.semaphore:
            try:
                start_time = asyncio.get_event_loop().time()
                result = await self.session.call_tool(tool_name, arguments)
                duration = asyncio.get_event_loop().time() - start_time
                
                return {
                    "tool": tool_name,
                    "success": True,
                    "result": result,
                    "duration": duration
                }
            except Exception as e:
                return {
                    "tool": tool_name,
                    "success": False,
                    "error":str(e),
                    "duration": 0
                }
    
    async def execute_batch(
        self, 
        tool_calls: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """Execute multiple tools in parallel"""
        tasks = [
            self.execute_tool(call["tool"], call["arguments"])
            for call in tool_calls
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        return [
            r if not isinstance(r, Exception) 
            else {"success": False, "error": str(r)}
            for r in results
        ]
    
    async def execute_with_dependencies(
        self,
        tool_graph: Dict[str, Dict]
    ) -> Dict[str, Any]:
        """Execute tools respecting dependencies"""
        completed = {}
        
        async def execute_node(node_id: str):
            node = tool_graph[node_id]
            
            # Wait for dependencies
            if "depends_on" in node:
                await asyncio.gather(*[
                    execute_node(dep) 
                    for dep in node["depends_on"]
                    if dep not in completed
                ])
            
            # Execute this node
            result = await self.execute_tool(
                node["tool"],
                node["arguments"]
            )
            completed[node_id] = result
            return result
        
        # Execute all root nodes in parallel
        root_nodes = [
            node_id for node_id, node in tool_graph.items()
            if "depends_on" not in node or not node["depends_on"]
        ]
        
        await asyncio.gather(*[
            execute_node(node_id) for node_id in root_nodes
        ])
        
        return completed

# Usage example
executor = ParallelMCPExecutor(mcp_session, max_concurrent=5)

# Batch execution
results = await executor.execute_batch([
    {"tool": "fetch_user_data", "arguments": {"user_id": "123"}},
    {"tool": "fetch_products", "arguments": {"category": "electronics"}},
    {"tool": "fetch_orders", "arguments": {"date_range": "last_7_days"}}
])

# Dependency-aware execution
tool_graph = {
    "fetch_user": {
        "tool": "get_user",
        "arguments": {"id": "123"}
    },
    "fetch_preferences": {
        "tool": "get_preferences",
        "arguments": {"user_id": "123"},
        "depends_on": ["fetch_user"]
    },
    "generate_recommendations": {
        "tool": "recommend",
        "arguments": {},
        "depends_on": ["fetch_user", "fetch_preferences"]
    }
}

results = await executor.execute_with_dependencies(tool_graph)

from typing import AsyncIterator
import json

class StreamingMCPTool:
    def __init__(self, session, tool_name: str):
        self.session = session
        self.tool_name = tool_name
    
    async def execute_streaming(
        self,
        arguments: dict,
        chunk_size: int = 1024
    ) -> AsyncIterator[dict]:
        """Execute tool with streaming response"""
        
        # Initiate the tool call
        response_stream = await self.session.call_tool_streaming(
            self.tool_name,
            arguments
        )
        
        buffer = ""
        
        async for chunk in response_stream:
            if isinstance(chunk, bytes):
                buffer += chunk.decode('utf-8')
            else:
                buffer += str(chunk)
            
            # Yield complete JSON objects from buffer
            while '\n' in buffer:
                line, buffer = buffer.split('\n', 1)
                if line.strip():
                    try:
                        yield json.loads(line)
                    except json.JSONDecodeError:
                        continue
        
        # Process any remaining data
        if buffer.strip():
            try:
                yield json.loads(buffer)
            except json.JSONDecodeError:
                pass

# Integration with LangChain agent
class StreamingAgent:
    def __init__(self, tools: List[StreamingMCPTool], llm):
        self.tools = tools
        self.llm = llm
    
    async def run_streaming(
        self,
        query: str
    ) -> AsyncIterator[dict]:
        """Run agent with streaming output"""
        
        # Get agent plan
        plan = await self.llm.aplan(query)
        
        for step in plan.steps:
            yield {
                "type": "step_start",
                "step": step.description
            }
            
            tool = next(
                t for t in self.tools 
                if t.tool_name == step.tool
            )
            
            async for chunk in tool.execute_streaming(step.arguments):
                yield {
                    "type": "tool_output",
                    "tool": step.tool,
                    "chunk": chunk
                }
            
            yield {
                "type": "step_complete",
                "step": step.description
            }

# Usage in web application
async def handle_streaming_request(query: str):
    agent = StreamingAgent(tools, llm)
    
    async for event in agent.run_streaming(query):
        # Send event to client (e.g., via WebSocket or SSE)
        await send_to_client(json.dumps(event))
        
        # Optional: Add delay for rate limiting
        await asyncio.sleep(0.01)

import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from your_module import MCPLangChainAdapter

class MockMCPSession:
    """Mock MCP session for testing"""
    
    def __init__(self):
        self.tools = [
            {
                "name": "test_tool",
                "description": "A test tool",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "input": {"type": "string"}
                    }
                }
            }
        ]
        self.call_count = 0
    
    async def initialize(self):
        return True
    
    async def list_tools(self):
        return self.tools
    
    async def call_tool(self, name, arguments):
        self.call_count += 1
        
        if name == "test_tool":
            return {
                "content": f"Processed: {arguments.get('input', '')}"
            }
        
        raise ValueError(f"Unknown tool: {name}")
    
    async def close(self):
        pass

@pytest.mark.asyncio
async def test_mcp_adapter_initialization():
    """Test adapter initialization"""
    mock_session = MockMCPSession()
    adapter = MCPLangChainAdapter(mock_session)
    
    tools = await adapter.get_langchain_tools()
    
    assert len(tools) == 1
    assert tools[0].name == "test_tool"

@pytest.mark.asyncio
async def test_tool_execution():
    """Test tool execution through adapter"""
    mock_session = MockMCPSession()
    adapter = MCPLangChainAdapter(mock_session)
    
    tools = await adapter.get_langchain_tools()
    tool = tools[0]
    
    result = await tool.arun(input="test data")
    
    assert "Processed: test data" in result
    assert mock_session.call_count == 1

@pytest.mark.asyncio
async def test_error_handling():
    """Test adapter error handling"""
    mock_session = MockMCPSession()
    adapter = MCPLangChainAdapter(mock_session)
    
    tools = await adapter.get_langchain_tools()
    tool = tools[0]
    
    # Simulate server error
    mock_session.call_tool = AsyncMock(
        side_effect=ConnectionError("Server unavailable")
    )
    
    with pytest.raises(ConnectionError):
        await tool.arun(input="test")

@pytest.mark.asyncio
async def test_retry_logic():
    """Test retry mechanism"""
    mock_session = MockMCPSession()
    
    # Fail twice, succeed third time
    call_count = 0
    async def flaky_call(name, args):
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise ConnectionError("Temporary failure")
        return {"content": "Success"}
    
    mock_session.call_tool = flaky_call
    
    adapter = MCPLangChainAdapter(mock_session, max_retries=3)
    tools = await adapter.get_langchain_tools()
    
    result = await tools[0].arun(input="test")
    assert "Success" in result
    assert call_count == 3

@pytest.mark.asyncio
async def test_caching_behavior():
    """Test response caching"""
    mock_session = MockMCPSession()
    adapter = MCPLangChainAdapter(mock_session, enable_cache=True)
    
    tools = await adapter.get_langchain_tools()
    tool = tools[0]
    
    # First call - cache miss
    result1 = await tool.arun(input="cached_test")
    assert mock_session.call_count == 1
    
    # Second call - cache hit
    result2 = await tool.arun(input="cached_test")
    assert mock_session.call_count == 1  # No additional call
    assert result1 == result2

# Integration test with real MCP server
@pytest.mark.integration
@pytest.mark.asyncio
async def test_real_mcp_integration():
    """Integration test with actual MCP server"""
    # This test requires a real MCP server running
    from mcp import ClientSession, StdioServerParameters
    
    server_params = StdioServerParameters(
        command="node",
        args=["./test-mcp-server/index.js"]
    )
    
    session = ClientSession(server_params)
    await session.initialize()
    
    adapter = MCPLangChainAdapter(session)
    tools = await adapter.get_langchain_tools()
    
    assert len(tools) > 0
    
    await session.close()