curl -LsSf https://astral.sh/uv/install.sh | sh

# 為我們的項(xiàng)目創(chuàng)建一個(gè)新目錄
uv init weather 
cd weather 
# 創(chuàng)建虛擬環(huán)境并激活它
uv venv 
source .venv/bin/activate 
# 安裝依賴項(xiàng)
uv add "mcp[cli]" httpx 
# 創(chuàng)建我們的服務(wù)器文件
touch weather.py

code .

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
#初始化FastMCP服務(wù)器
mcp = FastMCP("weather")
# 指定常量
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

async def make_nws_request(url: str) -> dict[str, Any] | None:
 """使用恰當(dāng)?shù)腻e(cuò)誤處理方式向NWS API提出請(qǐng)求。"""
 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 None
def format_alert(feature: dict) -> str:
 """將報(bào)警功能格式化為可讀的字符串。"""
 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:
 """獲取美國(guó)一個(gè)州的天氣警報(bào)。
 參數(shù):
 state: 兩個(gè)字母的美國(guó)州代碼（例如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:
 """獲取一個(gè)地點(diǎn)的天氣預(yù)報(bào)。
 參數(shù):
 latitude: 位置的緯度
 longitude: 位置的經(jīng)度
 """
 # 首先獲取預(yù)測(cè)網(wǎng)格端點(diǎn)
 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."
 # 從端點(diǎn)響應(yīng)中獲取預(yù)測(cè)的URL
 forecast_url = points_data["properties"]["forecast"]
 forecast_data = await make_nws_request(forecast_url)
 if not forecast_data:
 return "Unable to fetch detailed forecast."
 # 將時(shí)間范圍格式化為可讀的預(yù)測(cè)
 periods = forecast_data["properties"]["periods"]
 forecasts = []
 for period in periods[:5]: # 只顯示未來(lái)5個(gè)時(shí)段
 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__" : 
 # 初始化并運(yùn)行服務(wù)器
 mcp.run(transport= 'stdio' )

uv run weather.py

uv add ipykernel

uv run weather.py

!uv add langchain-mcp-adapters

!uv add langgraph langchain-anthropic python-dotenv

ANTHROPIC_API_KEY =sk-xxxxxxxx

from dotenv import load_dotenv
load_dotenv()
import os
api_key=os.environ.get("ANTHROPIC_API_KEY")

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-3-5-sonnet-20241022", api_key=api_key)

server_params = StdioServerParameters( 
 command= "python" , 
 # 確保更新為math_server.py文件的完整絕對(duì)路徑
 args=[ "./weather.py" ], 
) 

async with stdio_client(server_params) as (read, write): 
 async with ClientSession(read, write) as session: 
 # 初始化連接
 await session.initialize() 

 # 獲取工具
 tools = await load_mcp_tools(session) 

 # 創(chuàng)建并運(yùn)行代理
 agent = create_react_agent(model, tools) 
 agent_response = await agent.ainvoke({ "messages" : "加州目前的天氣怎么樣" })

agent_response

from IPython.display import display, Markdown
from langchain_core.messages import HumanMessage, ToolMessage, AIMessage
for response in agent_response["messages"]:
 user = ""

 if isinstance(response, HumanMessage):
 user = "**User**"
 elif isinstance(response, ToolMessage):
 user = "**Tool**"
 elif isinstance(response, AIMessage):
 user = "**AI**"

 if isinstance(response.content, list):
 display(Markdown(f'{user}: {response.content[0].get("text", "")}'))
 continue
 display(Markdown(f"{user}: {response.content}"))

if __name__ == "__main__" : 
 # 初始化并運(yùn)行服務(wù)器
 mcp.run(transport= 'sse' )

uv run weather.py

from langchain_mcp_adapters.client import MultiServerMCPClient

async with MultiServerMCPClient(
 {
 "weather": {
 "url": "http://localhost:8000/sse",
 "transport": "sse",
 }
 }
) as client:
 agent = create_react_agent(model, client.get_tools())
 agent_response = await agent.ainvoke({"messages": "what is the weather in nyc?"})

for response in agent_response["messages"]:
 user = ""

 if isinstance(response, HumanMessage):
 user = "**User**"
 elif isinstance(response, ToolMessage):
 user = "**Tool**"
 elif isinstance(response, AIMessage):
 user = "**AI**"

 if isinstance(response.content, list):
 display(Markdown(f'{user}: {response.content[0].get("text", "")}'))
 continue
 display(Markdown(f"{user}: {response.content}"))

async with MultiServerMCPClient( 
 { 
 "weather" : { 
 "url" : "http://localhost:8000/sse" , 
 "transport" : "sse" , 
 } 
 } 
) as client: 
 agent = create_react_agent(model, client.get_tools()) 

 # 流式傳輸響應(yīng)塊
 async for chunk in agent.astream({ "messages" : "what is the weather in nyc!" }): 
 # 從AddableUpdatesDict結(jié)構(gòu)中提取消息內(nèi)容
 if 'agent' in chunk and 'messages' in chunk[ 'agent' ]: 
 for message in chunk[ 'agent' ][ 'messages' ]: 
 if isinstance (message, AIMessage): 
 # 處理不同的內(nèi)容格式
 if isinstance (message.content, list ): 
 # 對(duì)于帶有文本和工具使用的結(jié)構(gòu)化內(nèi)容
 for item in message.content: 
 if isinstance (item, dict ) and 'text' in item: 
 display(Markdown( f"**AI**: {item[ 'text' ]} " )) 
 else : 
 # 對(duì)于簡(jiǎn)單文本內(nèi)容
 display(Markdown( 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' ): 
 # 顯示工具響應(yīng)
 display(Markdown( f"**Tool ( {message.name} )**: {message.content} " ))