Session 3: LangChain MCP Integration¶

# Basic MCP-LangChain integration

from langchain_mcp_adapters import MultiServerMCPClient
from langchain.agents import create_react_agent
from langchain_openai import ChatOpenAI

# Connect to multiple MCP servers

client = MultiServerMCPClient({
    "weather": {"command": "python", "args": ["weather_server.py"]},
    "files": {"command": "python", "args": ["file_server.py"]}
})

# Get tools from all servers

tools = client.list_tools()

# Create intelligent agent

agent = create_react_agent(
    llm=ChatOpenAI(model="gpt-4"),
    tools=tools,
    prompt="You are a helpful assistant with access to multiple tools."
)

# config.py - Foundation

import os
from typing import Dict, Any, List
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()

@dataclass
class MCPServerConfig:
    """Configuration for a single MCP server."""
    name: str
    command: str
    args: List[str]
    transport: str = "stdio"
    description: str = ""
    timeout: int = 30
    retry_attempts: int = 3

@dataclass 
class LLMConfig:
    """Configuration for language models."""
    provider: str = "openai"
    model: str = "gpt-4"
    temperature: float = 0.7
    max_tokens: int = 2000
    timeout: int = 60

# Main configuration class

class Config:
    """Main configuration class for LangChain MCP integration."""

    # API Keys from environment variables
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")

    # LLM Configuration with environment overrides
    LLM = LLMConfig(
        provider=os.getenv("LLM_PROVIDER", "openai"),
        model=os.getenv("LLM_MODEL", "gpt-4"),
        temperature=float(os.getenv("LLM_TEMPERATURE", "0.7")),
    )

    # MCP Server Registry
    MCP_SERVERS = [
        MCPServerConfig(
            name="weather",
            command="python",
            args=["mcp_servers/weather_server.py"],
            description="Weather information and forecasts"
        ),
        MCPServerConfig(
            name="filesystem", 
            command="python",
            args=["mcp_servers/filesystem_server.py"],
            description="Secure file system operations"
        ),
        MCPServerConfig(
            name="database",
            command="python", 
            args=["mcp_servers/database_server.py"],
            description="Database query and manipulation"
        )
    ]

    # Agent Configuration
    AGENT_CONFIG = {
        "max_iterations": int(os.getenv("MAX_ITERATIONS", "10")),
        "verbose": os.getenv("VERBOSE", "true").lower() == "true",
        "temperature": float(os.getenv("AGENT_TEMPERATURE", "0.7")),
        "timeout": int(os.getenv("AGENT_TIMEOUT", "300"))  # 5 minutes
    }

    # Logging Configuration
    LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
    LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

# Customer inquiry: "What's the weather like for my shipment to London?"

# Agent needs to:

# 1. Query customer database for shipment details

# 2. Get weather data for London

# 3. Check file system for shipping policies

# 4. Combine information into helpful response

# utils/mcp_manager.py - Foundation

import asyncio
import logging
from typing import Dict, List, Optional
from langchain_mcp_adapters import MCPAdapter
from config import Config, MCPServerConfig

logger = logging.getLogger(__name__)

class MCPServerManager:
    """Manages multiple MCP servers with health monitoring."""

    def __init__(self, server_configs: List[MCPServerConfig]):
        self.server_configs = {config.name: config for config in server_configs}
        self.adapters: Dict[str, MCPAdapter] = {}
        self.health_status: Dict[str, bool] = {}

    async def start_all_servers(self) -> Dict[str, bool]:
        """Start all configured MCP servers."""
        results = {}

        for name, config in self.server_configs.items():
            result = await self._start_single_server(name, config)
            results[name] = result

        return results

    async def _start_single_server(self, name: str, config: MCPServerConfig) -> bool:
        """Start a single MCP server and test its connection."""
        try:
            logger.info(f"Starting MCP server: {name}")
            adapter = MCPAdapter(
                command=config.command,
                args=config.args,
                timeout=config.timeout
            )

            # Test connection and discover tools
            await adapter.start()
            tools = await adapter.list_tools()

            # Store adapter and update status
            self.adapters[name] = adapter
            self.health_status[name] = True

            logger.info(f"Server '{name}' started with {len(tools)} tools")
            return True

        except Exception as e:
            logger.error(f"Failed to start server '{name}': {e}")
            self.health_status[name] = False
            return False

    async def get_adapter(self, server_name: str) -> Optional[MCPAdapter]:
        """Get adapter with automatic health checking."""
        if server_name not in self.adapters:
            logger.warning(f"Server '{server_name}' not found")
            return None

        if not self.health_status.get(server_name, False):
            logger.warning(f"Server '{server_name}' is unhealthy")
            # Attempt automatic restart
            await self._restart_server(server_name)

        return self.adapters.get(server_name)

    async def _restart_server(self, name: str):
        """Attempt to restart a failed server."""
        config = self.server_configs.get(name)
        if config:
            await self._start_single_server(name, config)

# Basic ReAct agent with MCP tools

from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# Simple ReAct prompt template

react_prompt = PromptTemplate.from_template("""
You have access to multiple tools. Use this format:

Question: {input}
Thought: I need to think about which tools to use
Action: [tool_name]
Action Input: [input_for_tool]
Observation: [result_from_tool]
... (repeat Thought/Action as needed)
Thought: I now have enough information
Final Answer: [comprehensive_response]

Available tools: {tools}
Question: {input}
{agent_scratchpad}
""")

# mcp_servers/weather_server.py

from mcp.server.fastmcp import FastMCP
from datetime import datetime
from typing import Dict

mcp = FastMCP("Weather Server")

# Sample weather data

WEATHER_DATA = {
    "London": {"temp": 15, "condition": "Cloudy", "humidity": 75},
    "New York": {"temp": 22, "condition": "Sunny", "humidity": 60},
    "Tokyo": {"temp": 18, "condition": "Rainy", "humidity": 85},
}

@mcp.tool()
def get_current_weather(city: str, units: str = "celsius") -> Dict:
    """Get current weather for a city."""
    if city not in WEATHER_DATA:
        return {"error": f"Weather data not available for {city}"}

    data = WEATHER_DATA[city].copy()
    if units == "fahrenheit":
        data["temp"] = (data["temp"] * 9/5) + 32
        data["units"] = "°F"
    else:
        data["units"] = "°C"

    data["city"] = city
    data["timestamp"] = datetime.now().isoformat()
    return data

if __name__ == "__main__":
    mcp.run()

# agents/basic_agent.py

import asyncio
import logging
from typing import Optional
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.tools import Tool

from utils.mcp_manager import MCPServerManager
from config import Config

logger = logging.getLogger(__name__)

class BasicMCPAgent:
    """A basic ReAct agent using a single MCP server."""

    def __init__(self, server_name: str, mcp_manager: MCPServerManager):
        self.server_name = server_name
        self.mcp_manager = mcp_manager
        self.llm = None
        self.agent_executor = None

    async def initialize(self) -> bool:
        """Initialize the agent with LLM and tools."""
        try:
            # Initialize LLM
            self.llm = ChatOpenAI(
                model=Config.LLM.model,
                temperature=Config.LLM.temperature,
                api_key=Config.OPENAI_API_KEY
            )

            # Get MCP adapter and tools
            adapter = await self.mcp_manager.get_adapter(self.server_name)
            if not adapter:
                logger.error(f"Failed to get adapter: {self.server_name}")
                return False

            # Convert MCP tools to LangChain tools
            mcp_tools = await adapter.list_tools()
            langchain_tools = self._create_langchain_tools(mcp_tools, adapter)

            # Create the agent executor
            self.agent_executor = self._create_agent_executor(langchain_tools)

            logger.info(f"Agent initialized with {len(langchain_tools)} tools")
            return True

        except Exception as e:
            logger.error(f"Failed to initialize agent: {e}")
            return False

    def _create_langchain_tools(self, mcp_tools, adapter):
        """Convert MCP tools to LangChain tools."""
        langchain_tools = []

        for mcp_tool in mcp_tools:
            # Create wrapper for MCP tool
            async def tool_wrapper(tool_input: str, tool_name=mcp_tool.name):
                """Wrapper to call MCP tool from LangChain."""
                try:
                    result = await adapter.call_tool(tool_name, {"input": tool_input})
                    return str(result)
                except Exception as e:
                    return f"Error calling tool {tool_name}: {str(e)}"

            langchain_tool = Tool(
                name=mcp_tool.name,
                description=mcp_tool.description or f"Tool from {self.server_name}",
                func=lambda x, tn=mcp_tool.name: asyncio.create_task(tool_wrapper(x, tn))
            )
            langchain_tools.append(langchain_tool)

        return langchain_tools

    async def run(self, query: str) -> str:
        """Run the agent with a query."""
        if not self.agent_executor:
            return "Agent not initialized. Call initialize() first."

        try:
            result = await self.agent_executor.ainvoke({"input": query})
            return result["output"]
        except Exception as e:
            return f"Error processing request: {str(e)}"

# agents/multi_tool_agent.py

import asyncio
import logging
from typing import Dict, List, Any, Optional
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.tools import Tool
from langchain.memory import ConversationBufferWindowMemory

from utils.mcp_manager import MCPServerManager
from config import Config

logger = logging.getLogger(__name__)

class MultiToolMCPAgent:
    """Advanced ReAct agent using multiple MCP servers."""

    def __init__(self, mcp_manager: MCPServerManager):
        self.mcp_manager = mcp_manager
        self.llm = None
        self.agent_executor = None
        self.memory = None
        self.available_tools = {}

    async def initialize(self) -> bool:
        """Initialize agent with LLM, tools, and memory."""
        try:
            # Initialize LLM
            self.llm = ChatOpenAI(
                model=Config.LLM.model,
                temperature=Config.LLM.temperature,
                api_key=Config.OPENAI_API_KEY
            )

            # Initialize conversation memory
            self.memory = ConversationBufferWindowMemory(
                k=10,  # Remember last 10 exchanges
                memory_key="chat_history",
                return_messages=True
            )

            # Collect tools from all available servers
            langchain_tools = await self._collect_all_tools()

            if not langchain_tools:
                logger.error("No tools available from MCP servers")
                return False

            # Create enhanced agent executor
            self.agent_executor = self._create_enhanced_agent(langchain_tools)

            tool_count = len(langchain_tools)
            server_count = len(self.available_tools)
            logger.info(f"Multi-tool agent: {tool_count} tools from {server_count} servers")

            return True

        except Exception as e:
            logger.error(f"Failed to initialize multi-tool agent: {e}")
            return False

    def _create_enhanced_agent(self, langchain_tools):
        """Create enhanced ReAct agent with intelligent prompting."""
        react_prompt = PromptTemplate.from_template("""
You are an intelligent AI assistant with access to multiple specialized tools.
You can use weather information, file system operations, and database queries.

INSTRUCTIONS:
1. Analyze the user's request carefully
2. Identify which tools might be helpful
3. Use tools in logical sequence
4. Provide comprehensive, helpful responses
5. If a tool fails, try alternative approaches

Available tools: {tools}

Use this format:
Question: {input}
Thought: Let me analyze what tools I need
Action: [tool_name]
Action Input: [input_for_tool]
Observation: [result_from_tool]
... (repeat as needed)
Thought: I have enough information
Final Answer: [comprehensive_response]

Conversation history: {chat_history}
Question: {input}
{agent_scratchpad}
""")

        agent = create_react_agent(
            llm=self.llm,
            tools=langchain_tools,
            prompt=react_prompt
        )

        return AgentExecutor(
            agent=agent,
            tools=langchain_tools,
            memory=self.memory,
            verbose=Config.AGENT_CONFIG["verbose"],
            max_iterations=Config.AGENT_CONFIG["max_iterations"],
            handle_parsing_errors=True,
            return_intermediate_steps=True
        )

# workflows/research_workflow.py

import asyncio
from typing import Dict, Any, List
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from dataclasses import dataclass

from utils.mcp_manager import MCPServerManager
from config import Config

@dataclass
class ResearchState:
    """State tracking data through workflow steps."""
    query: str
    messages: List[Any] 
    research_plan: str = ""
    weather_data: Dict = None
    file_data: Dict = None
    database_data: Dict = None
    final_report: str = ""
    step_count: int = 0

class ResearchWorkflow:
    """Advanced research workflow using LangGraph and multiple MCP servers."""

    def __init__(self, mcp_manager: MCPServerManager):
        self.mcp_manager = mcp_manager
        self.workflow = None

    async def build_workflow(self) -> StateGraph:
        """Build the LangGraph workflow graph."""
        workflow = StateGraph(ResearchState)

        # Add processing nodes
        workflow.add_node("planner", self._planning_node)
        workflow.add_node("weather_researcher", self._weather_research_node)
        workflow.add_node("file_researcher", self._file_research_node)
        workflow.add_node("database_researcher", self._database_research_node)
        workflow.add_node("synthesizer", self._synthesis_node)

        # Define execution flow
        workflow.set_entry_point("planner")
        workflow.add_edge("planner", "weather_researcher")
        workflow.add_edge("weather_researcher", "file_researcher")
        workflow.add_edge("file_researcher", "database_researcher")
        workflow.add_edge("database_researcher", "synthesizer")
        workflow.add_edge("synthesizer", END)

        self.workflow = workflow.compile()
        return self.workflow

    async def _planning_node(self, state: ResearchState) -> ResearchState:
        """Plan research approach based on query analysis."""
        query_lower = state.query.lower()
        plan_elements = []

        # Analyze query for different research domains
        if any(word in query_lower for word in ["weather", "climate", "temperature"]):
            plan_elements.append("- Gather weather information")

        if any(word in query_lower for word in ["file", "document", "data"]):
            plan_elements.append("- Search relevant files")

        if any(word in query_lower for word in ["database", "record", "history"]):
            plan_elements.append("- Query database for information")

        # Build research plan
        state.research_plan = "Research Plan:\n" + "\n".join(plan_elements) if plan_elements else "General research"
        state.step_count += 1
        return state

    async def _weather_research_node(self, state: ResearchState) -> ResearchState:
        """Research weather information if relevant."""
        if "weather" not in state.query.lower():
            state.weather_data = {"skipped": True}
            return state

        try:
            adapter = await self.mcp_manager.get_adapter("weather")
            if adapter:
                cities = self._extract_cities_from_query(state.query)
                weather_results = {}

                for city in cities:
                    try:
                        result = await adapter.call_tool("get_current_weather", {"city": city})
                        weather_results[city] = result
                    except:
                        continue  # Try other cities

                state.weather_data = weather_results or {"error": "No weather data"}
            else:
                state.weather_data = {"error": "Weather server unavailable"}

        except Exception as e:
            state.weather_data = {"error": str(e)}

        state.step_count += 1
        return state

    async def run_research(self, query: str) -> Dict[str, Any]:
        """Execute the complete research workflow."""
        if not self.workflow:
            await self.build_workflow()

        initial_state = ResearchState(query=query, messages=[HumanMessage(content=query)])

        try:
            final_state = await self.workflow.ainvoke(initial_state)
            return {
                "success": True,
                "query": query,
                "report": final_state.final_report,
                "steps": final_state.step_count
            }
        except Exception as e:
            return {
                "success": False,
                "query": query,
                "error": str(e),
                "report": f"Research workflow failed: {str(e)}"
            }

from langchain_mcp import MCPToolkit
from langchain.agents import initialize_agent, AgentType
from langchain.llm import OpenAI

# Your multi-server agent setup

weather_toolkit = MCPToolkit.from_server("weather-server")
file_toolkit = MCPToolkit.from_server("filesystem-server") 
preference_toolkit = MCPToolkit.from_server("preference-server")

# Combine toolkits and create agent

all_tools = weather_toolkit.get_tools() + file_toolkit.get_tools() + preference_toolkit.get_tools()