技术详解丨深度分析MCP工作原理，附代码实现（下）

# Hypothetical MCP stateful workflow (from official docs)class TravelAgent(MCPAgent):def __init__(self, user_id):self.context = load_context(user_id)# Load past interactionsself.preferences = self.context.get("preferences", {})
def book_flight(self, query):if self.preferences.get("class") == "economy":return search_flights(query, budget=True)else:return search_flights(query)

enum ErrorCode {// Standard JSON-RPC error codesParseError = -32700,InvalidRequest = -32600,MethodNotFound = -32601,InvalidParams = -32602,InternalError = -32603}

pip install mcp httpx langchain langchain-core langchai-community langchain-groq langchain-ollama langchain_mcp_adapters

import osfrom dotenv import load_dotenvload_dotenv()

# math_server.pyfrom mcp.server.fastmcp import FastMCP
mcp = FastMCP("Math")
@mcp.tool()def add(a: int, b: int) -> int:    """Add two numbers"""    return a + b
@mcp.tool()def multiply(a: int, b: int) -> int:    """Multiply two numbers"""    return a * b
if __name__ == "__main__":    mcp.run(transport="stdio")

from typing import Anyimport httpxfrom mcp.server.fastmcp import FastMCP# Initialize FastMCP servermcp = FastMCP("weather")# ConstantsNWS_API_BASE = "https://api.weather.gov"USER_AGENT = "weather-app/1.0"async def make_nws_request(url: str) -> dict[str, Any] | None:    """Make a request to the NWS API with proper error handling."""    headers = {        "User-Agent": USER_AGENT,        "Accept": "application/geo+json"    }    async with httpx.AsyncClient() as client:        try:            response = await client.get(url, headers=headers, timeout=30.0)            response.raise_for_status()            return response.json()        except Exception:            return Nonedef format_alert(feature: dict) -> str:    """Format an alert feature into a readable string."""    props = feature["properties"]    return f"""Event: {props.get('event', 'Unknown')}Area: {props.get('areaDesc', 'Unknown')}Severity: {props.get('severity', 'Unknown')}Description: {props.get('description', 'No description available')}Instructions: {props.get('instruction', 'No specific instructions provided')}"""@mcp.tool()async def get_alerts(state: str) -> str:    """Get weather alerts for a US state.    Args:        state: Two-letter US state code (e.g. CA, NY)    """    url = f"{NWS_API_BASE}/alerts/active/area/{state}"    data = await make_nws_request(url)    if not data or "features" not in data:        return "Unable to fetch alerts or no alerts found."    if not data["features"]:        return "No active alerts for this state."    alerts = [format_alert(feature) for feature in data["features"]]    return "\n---\n".join(alerts)@mcp.tool()async def get_forecast(latitude: float, longitude: float) -> str:    """Get weather forecast for a location.    Args:        latitude: Latitude of the location        longitude: Longitude of the location    """    # First get the forecast grid endpoint    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"    points_data = await make_nws_request(points_url)    if not points_data:        return "Unable to fetch forecast data for this location."    # Get the forecast URL from the points response    forecast_url = points_data["properties"]["forecast"]    forecast_data = await make_nws_request(forecast_url)    if not forecast_data:        return "Unable to fetch detailed forecast."    # Format the periods into a readable forecast    periods = forecast_data["properties"]["periods"]    forecasts = []    for period in periods[:5]:  # Only show next 5 periods        forecast = f"""{period['name']}:Temperature: {period['temperature']}°{period['temperatureUnit']}Wind: {period['windSpeed']} {period['windDirection']}Forecast: {period['detailedForecast']}"""        forecasts.append(forecast)    return "\n---\n".join(forecasts)if __name__ == "__main__":    # Initialize and run the server    mcp.run(transport='sse')

import asynciofrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom IPython.display import display, Markdownfrom langchain_core.messages import HumanMessage, ToolMessage, AIMessagefrom langchain_mcp_adapters.tools import load_mcp_toolsfrom langgraph.prebuilt import create_react_agentfrom langchain_mcp_adapters.client import MultiServerMCPClientfrom langgraph.prebuilt import create_react_agent#from langchain_azure_ai.chat_models import AzureAIChatCompletionsModelfrom langchain_groq import ChatGroqfrom langchain_ollama import ChatOllamafrom langchain_core.prompts import PromptTemplatefrom dotenv import load_dotenv
load_dotenv()
#model = ChatGroq(model="llama-3.3-70b-versatile",temperature=0.5)model = ChatOllama(model="llama3.2:1b",temperature=0.0,max_new_tokens=500)server_params = StdioServerParameters(    command="python",    # Make sure to update to the full absolute path to your math_server.py file    args=["weather.py"],)async def run_app(user_question):            async with MultiServerMCPClient(        {            "weather": {                "url": "http://localhost:8000/sse",                "transport": "sse",            },            "math": {            "command": "python",            # Make sure to update to the full absolute path to your math_server.py file            "args": ["math_server.py"],            "transport": "stdio",            },        }    ) as client:        agent = create_react_agent(model, client.get_tools())        agent_response = await agent.ainvoke({"messages": user_question})        print(agent_response['messages'][-1].content)        # # Stream the response chunks        # async for chunk in agent.astream({"messages": user_question}):        #     # Extract the message content from the AddableUpdatesDict structure        #     if 'agent' in chunk and 'messages' in chunk['agent']:        #         for message in chunk['agent']['messages']:        #             if isinstance(message, AIMessage):        #                 # Handle different content formats        #                 if isinstance(message.content, list):        #                     # For structured content with text and tool use        #                     for item in message.content:        #                         if isinstance(item, dict) and 'text' in item:        #                             print(f"**AI**: {item['text']}")        #                 else:        #                     # For simple text content        #                     print(f"**AI**: {message.content}")                                    #     elif 'tools' in chunk and 'messages' in chunk['tools']:        #         for message in chunk['tools']['messages']:        #             if hasattr(message, 'name') and hasattr(message, 'content'):        #                 # Display tool response        #                 print(f"**Tool ({message.name})**: {message.content}")        return agent_response['messages'][-1].contentif __name__ == "__main__":    #user_question = "what is the weather in california?"    #user_question = "what's (3 + 5) x 12?"    #user_question = "what's the weather in seattle?"    user_question = "what's the weather in NYC?"    response = asyncio.run(run_app(user_question=user_question))    print(response)

python weather.py

python langchain_mcp_multiserver.py

The result of (3 + 5) is 8, and 8 x 12 is 96.

It appears you've provided a list of weather alerts from the National Weather Service (NWS) for various regions in New York State, Vermont, and parts of Massachusetts.
Here's a breakdown of what each alert is saying:**Flooding Alerts*** The NWS has issued several flood watches across New York State, including:        + Northern St. Lawrence; Northern Franklin; Eastern Clinton; Southeastern St. Lawrence; Southern Franklin; Western Clinton; Western Essex; Southwestern St. Lawrence; Grand Isle; Western Franklin; Orleans; Essex; Western Chittenden; Lamoille; Caledonia; Washington; Western Addison; Orange; Western Rutland; Eastern Franklin; Eastern Chittenden; Eastern Addison; Eastern Rutland; Western Windsor; Eastern Windsor        + Northern Herkimer; Hamilton; Southern Herkimer; Southern Fulton; Montgomery; Northern Saratoga; Northern Warren; Northern Washington; Northern Fulton; Southeast Warren; Southern Washington; Bennington; Western Windham; Eastern Windham* The NWS has also issued a flood watch for parts of Vermont, including:        + Northern New York and northern and central Vermont**Ice Jam Alerts*** The NWS has warned about the possibility of ice jams in several areas, including:        + Bennington; Western Windham; Eastern Windham        + Southern Vermont, Bennington and Windham Counties        + Central New York, Herkimer County        + Northern New York, Hamilton, Montgomery, Fulton, Herkimer, Warren, Washington Counties**Other Alerts*** The NWS has issued several warnings about heavy rainfall and snowmelt leading to minor river flooding.* There are also alerts for isolated ice jams that could further increase the flood risk.It's essential to stay informed about weather conditions in your area and follow the instructions of local authorities. If you're planning outdoor activities, be prepared for changing weather conditions and take necessary precautions to stay safe.It appears you've provided a list of weather alerts from the National Weather Service (NWS) for various regions in New York State, Vermont, and parts of Massachusetts.Here's a breakdown of what each alert is saying:**Flooding Alerts*** The NWS has issued several flood watches across New York State, including:        + Northern St. Lawrence; Northern Franklin; Eastern Clinton; Southeastern St. Lawrence; Southern Franklin; Western Clinton; Western Essex; Southwestern St. Lawrence; Grand Isle; Western Franklin; Orleans; Essex; Western Chittenden; Lamoille; Caledonia; Washington; Western Addison; Orange; Western Rutland; Eastern Franklin; Eastern Chittenden; Eastern Addison; Eastern Rutland; Western Windsor; Eastern Windsor        + Northern Herkimer; Hamilton; Southern Herkimer; Southern Fulton; Montgomery; Northern Saratoga; Northern Warren; Northern Washington; Northern Fulton; Southeast Warren; Southern Washington; Bennington; Western Windham; Eastern Windham* The NWS has also issued a flood watch for parts of Vermont, including:        + Northern New York and northern and central Vermont**Ice Jam Alerts*** The NWS has warned about the possibility of ice jams in several areas, including:        + Bennington; Western Windham; Eastern Windham        + Southern Vermont, Bennington and Windham Counties        + Central New York, Herkimer County        + Northern New York, Hamilton, Montgomery, Fulton, Herkimer, Warren, Washington Counties**Other Alerts*** The NWS has issued several warnings about heavy rainfall and snowmelt leading to minor river flooding.* There are also alerts for isolated ice jams that could further increase the flood risk.
It's essential to stay informed about weather conditions in your area and follow the instructions of local authorities. If you're planning outdoor activities, be prepared for changing weather conditions and take necessary precautions to stay safe.