Model Context Protocol (MCP) \u662f\u4e00\u4e2a\u5f3a\u5927\u7684\u6846\u67b6\uff0c\u5141\u8bb8 AI \u6a21\u578b\u901a\u8fc7\u6807\u51c6\u5316\u7684\u534f\u8bae\u8c03\u7528\u5916\u90e8\u5de5\u5177\u548c\u670d\u52a1\u3002\u672c\u6587\u5c06\u901a\u8fc7\u4e00\u4e2a\u5b9e\u9645\u6848\u4f8b\u2014\u2014\u667a\u80fd\u65b0\u95fb\u5206\u6790\u4e0e\u90ae\u4ef6\u53d1\u9001\u7cfb\u7edf<\/strong>\uff0c\u6df1\u5165\u7406\u89e3\u5982\u4f55\u5f00\u53d1\u4e00\u4e2a\u5b8c\u6574\u7684 MCP \u670d\u52a1\u3002<\/p>\n\n\n\n \u6211\u4eec\u5c06\u6784\u5efa\u4e00\u4e2a\u7cfb\u7edf\uff0c\u5b83\u80fd\u591f\uff1a<\/p>\n\n\n\n MCP Server \u8d1f\u8d23\u5b9a\u4e49\u548c\u5b9e\u73b0\u5177\u4f53\u7684\u5de5\u5177\u529f\u80fd\u3002<\/p>\n\n\n\n \u4f7f\u7528 MCP Client \u8d1f\u8d23\u4e0e AI \u6a21\u578b\u4ea4\u4e92\uff0c\u89c4\u5212\u5de5\u5177\u8c03\u7528\u94fe\u3002<\/p>\n\n\n\n \u8fd9\u662f MCP Client \u7684\u6838\u5fc3\u529f\u80fd\u2014\u2014\u8ba9 AI \u6a21\u578b\u89c4\u5212\u5de5\u5177\u8c03\u7528\u94fe\uff1a<\/p>\n\n\n\n \u5173\u952e\u70b9<\/strong>\uff1a\u901a\u8fc7\u7cbe\u5fc3\u8bbe\u8ba1\u7684 system prompt\uff0c\u8ba9 AI \u7406\u89e3\uff1a<\/p>\n\n\n\n\ud83c\udfaf \u9879\u76ee\u6982\u8ff0<\/h2>\n\n\n\n
\n
\u6280\u672f\u6808<\/h3>\n\n\n\n
\n
mcp<\/code> \u548c fastmcp<\/code><\/li>\n\n\n\n
\n\n\n\n\ud83c\udfd7\ufe0f \u67b6\u6784\u8bbe\u8ba1<\/h2>\n\n\n\n
\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510
\u2502 \u7528\u6237\u8f93\u5165 \u2502 \u2500\u2500\u25b6 \u2502 MCP Client \u2502 \u2500\u2500\u25b6 \u2502 MCP Server \u2502
\u2502 \u81ea\u7136\u8bed\u8a00 \u2502 \u2502 (client.py) \u2502 \u2502 (server.py) \u2502
\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518
\u2502 \u2502
\u2502 \u251c\u2500\u25b6 \ud83d\udd0d \u641c\u7d22\u65b0\u95fb
\u2502 \u251c\u2500\u25b6 \ud83e\udde0 \u60c5\u611f\u5206\u6790
\u25bc \u2514\u2500\u25b6 \ud83d\udce7 \u53d1\u9001\u90ae\u4ef6
\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510
\u2502 AI \u6a21\u578b \u2502
\u2502 (Qwen) \u2502
\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518<\/pre>\n\n\n\n
\n\n\n\n\ud83d\udcdd \u7b2c\u4e00\u6b65\uff1a\u521b\u5efa MCP Server<\/h2>\n\n\n\n
1.1 \u521d\u59cb\u5316\u670d\u52a1\u5668<\/h3>\n\n\n\n
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
\u200b
# \u52a0\u8f7d\u73af\u5883\u53d8\u91cf
load_dotenv()
\u200b
# \u521d\u59cb\u5316 MCP \u670d\u52a1\u5668
mcp = FastMCP(\"NewsServer\")<\/pre>\n\n\n\n1.2 \u5b9a\u4e49\u5de5\u5177\uff1a\u641c\u7d22\u65b0\u95fb<\/h3>\n\n\n\n
@mcp.tool()<\/code> \u88c5\u9970\u5668\u5b9a\u4e49\u5de5\u5177\uff1a<\/p>\n\n\n\nimport httpx
import json
import os
from datetime import datetime
\u200b
@mcp.tool()
async def search_google_news(keyword: str) -> str:
\"\"\"
\u4f7f\u7528 Serper API \u641c\u7d22\u65b0\u95fb\u5185\u5bb9\uff0c\u8fd4\u56de\u524d5\u6761\u6807\u9898\u3001\u63cf\u8ff0\u548c\u94fe\u63a5\u3002
\u53c2\u6570:
keyword (str): \u641c\u7d22\u5173\u952e\u8bcd\uff0c\u5982 \"\u5c0f\u7c73\u6c7d\u8f66\"
\u8fd4\u56de:
str: JSON \u5b57\u7b26\u4e32\uff0c\u5305\u542b\u65b0\u95fb\u6807\u9898\u3001\u63cf\u8ff0\u3001\u94fe\u63a5
\"\"\"
api_key = os.getenv(\"SERPER_API_KEY\")
if not api_key:
return \"\u274c \u672a\u914d\u7f6e SERPER_API_KEY\"
url = \"https:\/\/google.serper.dev\/news\"
headers = {
\"X-API-KEY\": api_key,
\"Content-Type\": \"application\/json\"
}
payload = {\"q\": keyword}
async with httpx.AsyncClient() as client:
response = await client.post(url, headers=headers, json=payload)
data = response.json()
if \"news\" not in data:
return \"\u274c \u672a\u83b7\u53d6\u5230\u641c\u7d22\u7ed3\u679c\"
# \u63d0\u53d6\u524d5\u6761\u65b0\u95fb
articles = [
{
\"title\": item.get(\"title\"),
\"desc\": item.get(\"snippet\"),
\"url\": item.get(\"link\")
} for item in data[\"news\"][:5]
]
# \u4fdd\u5b58\u5230\u672c\u5730
output_dir = \".\/google_news\"
os.makedirs(output_dir, exist_ok=True)
filename = f\"google_news_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json\"
file_path = os.path.join(output_dir, filename)
with open(file_path, \"w\", encoding=\"utf-8\") as f:
json.dump(articles, f, ensure_ascii=False, indent=2)
return (
f\"\u2705 \u5df2\u83b7\u53d6\u4e0e [{keyword}] \u76f8\u5173\u7684\u524d5\u6761 Google \u65b0\u95fb\uff1a\\n\"
f\"{json.dumps(articles, ensure_ascii=False, indent=2)}\\n\"
f\"\ud83d\udcc4 \u5df2\u4fdd\u5b58\u5230\uff1a{file_path}\"
)<\/pre>\n\n\n\n1.3 \u5b9a\u4e49\u5de5\u5177\uff1a\u60c5\u611f\u5206\u6790<\/h3>\n\n\n\n
from openai import OpenAI
\u200b
@mcp.tool()
async def analyze_sentiment(text: str, filename: str) -> str:
\"\"\"
\u5bf9\u6587\u672c\u8fdb\u884c\u60c5\u611f\u5206\u6790\uff0c\u5e76\u4fdd\u5b58\u4e3a Markdown \u6587\u4ef6\u3002
\u53c2\u6570:
text (str): \u65b0\u95fb\u63cf\u8ff0\u6216\u6587\u672c\u5185\u5bb9
filename (str): \u4fdd\u5b58\u7684 Markdown \u6587\u4ef6\u540d\uff08\u4e0d\u542b\u8def\u5f84\uff09
\u8fd4\u56de:
str: \u4ec5\u8fd4\u56de\u6587\u4ef6\u540d\uff08\u7528\u4e8e\u90ae\u4ef6\u53d1\u9001\uff09
\"\"\"
# \u521d\u59cb\u5316 AI \u5ba2\u6237\u7aef
openai_key = os.getenv(\"DASHSCOPE_API_KEY\")
model = os.getenv(\"MODEL\")
client = OpenAI(
api_key=openai_key,
base_url=os.getenv(\"BASE_URL\")
)
# \u6784\u9020\u63d0\u793a\u8bcd
prompt = f\"\u8bf7\u5bf9\u4ee5\u4e0b\u65b0\u95fb\u5185\u5bb9\u8fdb\u884c\u60c5\u7eea\u503e\u5411\u5206\u6790\uff0c\u5e76\u8bf4\u660e\u539f\u56e0\uff1a\\n\\n{text}\"
# \u8c03\u7528 AI \u6a21\u578b
response = client.chat.completions.create(
model=model,
messages=[{\"role\": \"user\", \"content\": prompt}]
)
result = response.choices[0].message.content.strip()
# \u751f\u6210 Markdown \u62a5\u544a
markdown = f\"\"\"# \u8206\u60c5\u5206\u6790\u62a5\u544a
\u200b
**\u5206\u6790\u65f6\u95f4\uff1a** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
\u200b
---
\u200b
## \ud83d\udce5 \u539f\u59cb\u6587\u672c
\u200b
{text}
\u200b
---
\u200b
## \ud83d\udcca \u5206\u6790\u7ed3\u679c
\u200b
{result}
\"\"\"
# \u4fdd\u5b58\u6587\u4ef6
output_dir = \".\/sentiment_reports\"
os.makedirs(output_dir, exist_ok=True)
if not filename:
filename = f\"sentiment_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md\"
file_path = os.path.join(output_dir, filename)
with open(file_path, \"w\", encoding=\"utf-8\") as f:
f.write(markdown)
# \u53ea\u8fd4\u56de\u6587\u4ef6\u540d\uff0c\u65b9\u4fbf\u540e\u7eed\u90ae\u4ef6\u53d1\u9001
return filename<\/pre>\n\n\n\n1.4 \u5b9a\u4e49\u5de5\u5177\uff1a\u53d1\u9001\u90ae\u4ef6<\/h3>\n\n\n\n
import smtplib
from email.message import EmailMessage
\u200b
@mcp.tool()
async def send_email_with_attachment(
to: str,
subject: str,
body: str,
filename: str
) -> str:
\"\"\"
\u53d1\u9001\u5e26\u9644\u4ef6\u7684\u90ae\u4ef6\u3002
\u53c2\u6570:
to: \u6536\u4ef6\u4eba\u90ae\u7bb1\u5730\u5740
subject: \u90ae\u4ef6\u6807\u9898
body: \u90ae\u4ef6\u6b63\u6587
filename: \u6587\u4ef6\u540d\uff08\u4e0d\u542b\u8def\u5f84\uff09
\u8fd4\u56de:
\u90ae\u4ef6\u53d1\u9001\u72b6\u6001\u8bf4\u660e
\"\"\"
# \u83b7\u53d6 SMTP \u914d\u7f6e
smtp_server = os.getenv(\"SMTP_SERVER\") # smtp.qq.com
smtp_port = int(os.getenv(\"SMTP_PORT\", 465))
sender_email = os.getenv(\"EMAIL_USER\")
sender_pass = os.getenv(\"EMAIL_PASS\") # QQ \u90ae\u7bb1\u6388\u6743\u7801
# \u83b7\u53d6\u9644\u4ef6\u6587\u4ef6\u8def\u5f84
full_path = os.path.abspath(
os.path.join(\".\/sentiment_reports\", filename)
)
if not os.path.exists(full_path):
return f\"\u274c \u9644\u4ef6\u8def\u5f84\u65e0\u6548\uff0c\u672a\u627e\u5230\u6587\u4ef6: {full_path}\"
# \u521b\u5efa\u90ae\u4ef6
msg = EmailMessage()
msg[\"Subject\"] = subject
msg[\"From\"] = sender_email
msg[\"To\"] = to
msg.set_content(body)
# \u6dfb\u52a0\u9644\u4ef6
try:
with open(full_path, \"rb\") as f:
file_data = f.read()
file_name = os.path.basename(full_path)
msg.add_attachment(
file_data,
maintype=\"application\",
subtype=\"octet-stream\",
filename=file_name
)
except Exception as e:
return f\"\u274c \u9644\u4ef6\u8bfb\u53d6\u5931\u8d25: {str(e)}\"
# \u53d1\u9001\u90ae\u4ef6
try:
with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
server.login(sender_email, sender_pass)
server.send_message(msg)
return f\"\u2705 \u90ae\u4ef6\u5df2\u6210\u529f\u53d1\u9001\u7ed9 {to}\uff0c\u9644\u4ef6: {full_path}\"
except Exception as e:
return f\"\u274c \u90ae\u4ef6\u53d1\u9001\u5931\u8d25: {str(e)}\"<\/pre>\n\n\n\n1.5 \u542f\u52a8\u670d\u52a1\u5668<\/h3>\n\n\n\n
if __name__ == \"__main__\":
mcp.run(transport='stdio')<\/pre>\n\n\n\n
\n\n\n\n\ud83e\udd16 \u7b2c\u4e8c\u6b65\uff1a\u521b\u5efa MCP Client<\/h2>\n\n\n\n
2.1 \u521d\u59cb\u5316\u5ba2\u6237\u7aef<\/h3>\n\n\n\n
import asyncio
import os
import json
from typing import Optional, List
from contextlib import AsyncExitStack
from datetime import datetime
import re
from openai import OpenAI
from dotenv import load_dotenv
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
\u200b
load_dotenv()
\u200b
class MCPClient:
def __init__(self):
self.exit_stack = AsyncExitStack()
self.openai_api_key = os.getenv(\"DASHSCOPE_API_KEY\")
self.base_url = os.getenv(\"BASE_URL\")
self.model = os.getenv(\"MODEL\")
if not self.openai_api_key:
raise ValueError(\"\u274c \u672a\u627e\u5230 API Key\")
self.client = OpenAI(
api_key=self.openai_api_key,
base_url=self.base_url
)
self.session: Optional[ClientSession] = None<\/pre>\n\n\n\n2.2 \u8fde\u63a5\u5230 MCP Server<\/h3>\n\n\n\n
async def connect_to_server(self, server_script_path: str):
\"\"\"\u8fde\u63a5\u5230 MCP \u670d\u52a1\u5668\"\"\"
# \u5224\u65ad\u670d\u52a1\u5668\u811a\u672c\u7c7b\u578b
is_python = server_script_path.endswith('.py')
is_js = server_script_path.endswith('.js')
if not (is_python or is_js):
raise ValueError(\"\u670d\u52a1\u5668\u811a\u672c\u5fc5\u987b\u662f .py \u6216 .js \u6587\u4ef6\")
# \u786e\u5b9a\u542f\u52a8\u547d\u4ee4
command = \"python\" if is_python else \"node\"
# \u6784\u9020\u670d\u52a1\u5668\u53c2\u6570
server_params = StdioServerParameters(
command=command,
args=[server_script_path],
env=None
)
# \u542f\u52a8\u670d\u52a1\u8fdb\u7a0b
stdio_transport = await self.exit_stack.enter_async_context(
stdio_client(server_params)
)
self.stdio, self.write = stdio_transport
# \u521b\u5efa\u4f1a\u8bdd
self.session = await self.exit_stack.enter_async_context(
ClientSession(self.stdio, self.write)
)
# \u521d\u59cb\u5316\u4f1a\u8bdd
await self.session.initialize()
# \u83b7\u53d6\u5de5\u5177\u5217\u8868
response = await self.session.list_tools()
tools = response.tools
print(\"\\n\u5df2\u8fde\u63a5\u5230\u670d\u52a1\u5668\uff0c\u652f\u6301\u4ee5\u4e0b\u5de5\u5177:\",
[tool.name for tool in tools])<\/pre>\n\n\n\n2.3 \u5de5\u5177\u8c03\u7528\u89c4\u5212\uff08\u6838\u5fc3\u903b\u8f91\uff09<\/h3>\n\n\n\n
async def plan_tool_usage(
self,
query: str,
tools: List[dict]
) -> List[dict]:
\"\"\"\u4f7f\u7528 AI \u89c4\u5212\u5de5\u5177\u8c03\u7528\u94fe\"\"\"
print(\"\\n\ud83d\udce4 \u63d0\u4ea4\u7ed9\u5927\u6a21\u578b\u7684\u5de5\u5177\u5b9a\u4e49:\")
print(json.dumps(tools, ensure_ascii=False, indent=2))
# \u6784\u9020\u5de5\u5177\u5217\u8868\u6587\u672c
tool_list_text = \"\\n\".join([
f\"- {tool['function']['name']}: \"
f\"{tool['function']['description']}\"
for tool in tools
])
# \u6784\u9020\u7cfb\u7edf\u63d0\u793a\u8bcd
system_prompt = {
\"role\": \"system\",
\"content\": (
\"\u4f60\u662f\u4e00\u4e2a\u667a\u80fd\u4efb\u52a1\u89c4\u5212\u52a9\u624b\uff0c\u7528\u6237\u4f1a\u7ed9\u51fa\u81ea\u7136\u8bed\u8a00\u8bf7\u6c42\u3002\\n\"
\"\u4f60\u53ea\u80fd\u4ece\u4ee5\u4e0b\u5de5\u5177\u4e2d\u9009\u62e9\uff08\u4e25\u683c\u4f7f\u7528\u5de5\u5177\u540d\u79f0\uff09\uff1a\\n\"
f\"{tool_list_text}\\n\\n\"
\"**\u91cd\u8981\u89c4\u5219**\uff1a\\n\"
\"1. \u5982\u679c\u7528\u6237\u8981\u6c42'\u641c\u7d22\u65b0\u95fb\u5e76\u5206\u6790\u5e76\u53d1\u9001\u90ae\u4ef6'\uff0c\"
\"\u4f60\u5fc5\u987b\u89c4\u52123\u4e2a\u6b65\u9aa4\uff1a\\n\"
\" \u6b65\u9aa41: search_google_news (\u641c\u7d22\u65b0\u95fb)\\n\"
\" \u6b65\u9aa42: analyze_sentiment (\u5206\u6790\u5e76\u751f\u6210\u6587\u4ef6)\\n\"
\" \u6b65\u9aa43: send_email_with_attachment (\u53d1\u9001\u90ae\u4ef6)\\n\"
\"2. \u540e\u7eed\u6b65\u9aa4\u53ef\u4ee5\u4f7f\u7528 {{\u5de5\u5177\u540d}} \u6765\u5f15\u7528\u4e0a\u4e00\u6b65\u7684\u8f93\u51fa\\n\"
\"3. analyze_sentiment \u8fd4\u56de\u7684\u662f\u6587\u4ef6\u540d\uff08\u4e0d\u542b\u8def\u5f84\uff09\uff0c\"
\"\u76f4\u63a5\u7528\u4e8e send_email_with_attachment \u7684 filename \u53c2\u6570\\n\"
\"4. \u8fd4\u56de\u683c\u5f0f\u5fc5\u987b\u662f\u7eaf JSON \u6570\u7ec4\uff0c\u4e0d\u8981\u6dfb\u52a0\u4efb\u4f55\u89e3\u91ca\\n\\n\"
\"\u793a\u4f8b\u8f93\u51fa\u683c\u5f0f\uff1a\\n\"
'[\\n'
' {\"name\": \"search_google_news\", '
'\"arguments\": {\"keyword\": \"\u5c0f\u7c73SU7\"}},\\n'
' {\"name\": \"analyze_sentiment\", '
'\"arguments\": {\"text\": \"{{search_google_news}}\", '
'\"filename\": \"report.md\"}},\\n'
' {\"name\": \"send_email_with_attachment\", '
'\"arguments\": {\"to\": \"user@qq.com\", '
'\"subject\": \"\u62a5\u544a\", \"body\": \"\u8bf7\u67e5\u6536\", '
'\"filename\": \"{{analyze_sentiment}}\"}}\\n'
']'
)
}
# \u8c03\u7528 AI \u6a21\u578b\u89c4\u5212
planning_messages = [
system_prompt,
{\"role\": \"user\", \"content\": query}
]
response = self.client.chat.completions.create(
model=self.model,
messages=planning_messages,
tools=tools,
tool_choice=\"none\"
)
# \u63d0\u53d6 JSON \u5185\u5bb9
content = response.choices[0].message.content.strip()
match = re.search(r\"```(?:json)?\\s*([\\s\\S]+?)\\s*```\", content)
if match:
json_text = match.group(1)
else:
json_text = content
# \u89e3\u6790\u5e76\u8fd4\u56de\u8ba1\u5212
try:
plan = json.loads(json_text)
print(f\"\\n\ud83d\udd27 AI \u751f\u6210\u7684\u5de5\u5177\u8c03\u7528\u8ba1\u5212:\")
print(json.dumps(plan, ensure_ascii=False, indent=2))
return plan if isinstance(plan, list) else []
except Exception as e:
print(f\"\u274c \u5de5\u5177\u8c03\u7528\u94fe\u89c4\u5212\u5931\u8d25: {e}\\n\u539f\u59cb\u8fd4\u56de: {content}\")
return []<\/pre>\n\n\n\n2.4 \u6267\u884c\u5de5\u5177\u8c03\u7528\u94fe<\/h3>\n\n\n\n
async def process_query(self, query: str) -> str:
\"\"\"\u5904\u7406\u7528\u6237\u67e5\u8be2\"\"\"
# \u83b7\u53d6\u53ef\u7528\u5de5\u5177
response = await self.session.list_tools()
available_tools = [
{
\"type\": \"function\",
\"function\": {
\"name\": tool.name,
\"description\": tool.description,
\"input_schema\": tool.inputSchema
}
} for tool in response.tools
]
# \u751f\u6210\u6587\u4ef6\u540d\uff08\u7528\u4e8e\u5206\u6790\u62a5\u544a\uff09
keyword_match = re.search(
r'(\u5173\u4e8e|\u5206\u6790|\u67e5\u8be2|\u641c\u7d22|\u67e5\u770b)([^\u7684\\s\uff0c\u3002\u3001\uff1f\\n]+)',
query
)
keyword = keyword_match.group(2) if keyword_match else \"\u5206\u6790\u5bf9\u8c61\"
safe_keyword = re.sub(r'[\\\\\/:*?\"<>|]', '', keyword)[:20]
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
md_filename = f\"sentiment_{safe_keyword}_{timestamp}.md\"
md_path = os.path.join(\".\/sentiment_reports\", md_filename)
# \u5c06\u6587\u4ef6\u540d\u4fe1\u606f\u6ce8\u5165\u67e5\u8be2
query_with_filename = (
f\"{query.strip()} \"
f\"[md_filename={md_filename}] [md_path={md_path}]\"
)
# \u89c4\u5212\u5de5\u5177\u8c03\u7528\u94fe
tool_plan = await self.plan_tool_usage(
query_with_filename,
available_tools
)
# \u6267\u884c\u5de5\u5177\u8c03\u7528
tool_outputs = {}
messages = [{\"role\": \"user\", \"content\": query_with_filename}]
for step in tool_plan:
tool_name = step[\"name\"]
tool_args = step[\"arguments\"]
# \u89e3\u6790\u5360\u4f4d\u7b26\uff08\u5982 {{search_google_news}}\uff09
for key, val in tool_args.items():
if isinstance(val, str) and \\
val.startswith(\"{{\") and val.endswith(\"}}\"):
ref_key = val.strip(\"{} \")
resolved_val = tool_outputs.get(ref_key, val)
tool_args[key] = resolved_val
# \u6ce8\u5165\u6587\u4ef6\u540d
if tool_name == \"analyze_sentiment\" and \\
\"filename\" not in tool_args:
tool_args[\"filename\"] = md_filename
if tool_name == \"send_email_with_attachment\" and \\
\"attachment_path\" not in tool_args:
tool_args[\"attachment_path\"] = md_path
# \u8c03\u7528\u5de5\u5177
result = await self.session.call_tool(tool_name, tool_args)
# \u4fdd\u5b58\u8f93\u51fa
tool_outputs[tool_name] = result.content[0].text
# \u5c06\u5de5\u5177\u8f93\u51fa\u6dfb\u52a0\u5230\u6d88\u606f\u6d41
# \u6ce8\u610f\uff1a\u4f7f\u7528 assistant \u89d2\u8272\u800c\u975e tool\uff0c\u907f\u514d API \u9650\u5236
messages.append({
\"role\": \"assistant\",
\"content\": f\"[tool:{tool_name} \u8f93\u51fa]\\n{result.content[0].text}\"
})
# \u751f\u6210\u6700\u7ec8\u56de\u590d
final_response = self.client.chat.completions.create(
model=self.model,
messages=messages
)
final_output = final_response.choices[0].message.content
# \u4fdd\u5b58\u5bf9\u8bdd\u8bb0\u5f55
def clean_filename(text: str) -> str:
text = text.strip()
text = re.sub(r'[\\\\\/:*?\"<>|]', '', text)
return text[:50]
safe_filename = clean_filename(query)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f\"{safe_filename}_{timestamp}.txt\"
output_dir = \".\/llm_outputs\"
os.makedirs(output_dir, exist_ok=True)
file_path = os.path.join(output_dir, filename)
with open(file_path, \"w\", encoding=\"utf-8\") as f:
f.write(f\"\ud83d\udde3 \u7528\u6237\u63d0\u95ee\uff1a{query}\\n\\n\")
f.write(f\"\ud83e\udd16 \u6a21\u578b\u56de\u590d\uff1a\\n{final_output}\\n\")
print(f\"\ud83d\udcc4 \u5bf9\u8bdd\u8bb0\u5f55\u5df2\u4fdd\u5b58\u4e3a\uff1a{file_path}\")
return final_output<\/pre>\n\n\n\n2.5 \u4ea4\u4e92\u5faa\u73af<\/h3>\n\n\n\n
async def chat_loop(self):
\"\"\"\u4e3b\u4ea4\u4e92\u5faa\u73af\"\"\"
print(\"\\n\ud83e\udd16 MCP \u5ba2\u6237\u7aef\u5df2\u542f\u52a8\uff01\u8f93\u5165 'quit' \u9000\u51fa\")
while True:
try:
query = input(\"\\n\u4f60: \").strip()
if query.lower() == 'quit':
break
response = await self.process_query(query)
print(f\"\\n\ud83e\udd16 AI: {response}\")
except Exception as e:
print(f\"\\n\u26a0\ufe0f \u53d1\u751f\u9519\u8bef: {str(e)}\")
async def cleanup(self):
\"\"\"\u6e05\u7406\u8d44\u6e90\"\"\"
await self.exit_stack.aclose()<\/pre>\n\n\n\n2.6 \u4e3b\u5165\u53e3<\/h3>\n\n\n\n
async def main():
server_script_path = os.path.join(
os.path.dirname(__file__),
\"server.py\"
)
client = MCPClient()
try:
await client.connect_to_server(server_script_path)
await client.chat_loop()
finally:
await client.cleanup()
\u200b
if __name__ == \"__main__\":
asyncio.run(main())<\/pre>\n\n\n\n
\n\n\n\n\u2699\ufe0f \u7b2c\u4e09\u6b65\uff1a\u73af\u5883\u914d\u7f6e<\/h2>\n\n\n\n
3.1 \u521b\u5efa
.env<\/code> \u6587\u4ef6<\/h3>\n\n\n\n# AI \u6a21\u578b\u914d\u7f6e\uff08\u901a\u4e49\u5343\u95ee\uff09
BASE_URL=https:\/\/dashscope.aliyuncs.com\/compatible-mode\/v1
MODEL=qwen-plus
DASHSCOPE_API_KEY=sk-your-api-key-here
\u200b
# \u65b0\u95fb\u641c\u7d22 API
SERPER_API_KEY=your-serper-api-key
\u200b
# QQ \u90ae\u7bb1 SMTP \u914d\u7f6e
SMTP_SERVER=smtp.qq.com
SMTP_PORT=465
EMAIL_USER=your-email@qq.com
EMAIL_PASS=your-authorization-code<\/pre>\n\n\n\n3.2 \u83b7\u53d6 QQ \u90ae\u7bb1\u6388\u6743\u7801<\/h3>\n\n\n\n
\n
EMAIL_PASS<\/code><\/li>\n<\/ol>\n\n\n\n3.3 \u5b89\u88c5\u4f9d\u8d56<\/h3>\n\n\n\n
pip install mcp openai python-dotenv httpx<\/pre>\n\n\n\n
\n\n\n\n\ud83d\ude80 \u7b2c\u56db\u6b65\uff1a\u8fd0\u884c\u7cfb\u7edf<\/h2>\n\n\n\n
4.1 \u542f\u52a8\u5ba2\u6237\u7aef<\/h3>\n\n\n\n
python client.py<\/pre>\n\n\n\n
4.2 \u4f7f\u7528\u793a\u4f8b<\/h3>\n\n\n\n
\u4f60: \u5206\u6790\u4e00\u4e0b\u5c0f\u7c73SU7\u8fd1\u671f\u7684\u70ed\u70b9\u65b0\u95fb\uff0c\u5e76\u53d1\u9001\u90ae\u4ef6\u7ed9 user@qq.com<\/pre>\n\n\n\n
4.3 \u6267\u884c\u6d41\u7a0b<\/h3>\n\n\n\n
\ud83d\udce4 \u63d0\u4ea4\u7ed9\u5927\u6a21\u578b\u7684\u5de5\u5177\u5b9a\u4e49
\ud83d\udd27 AI \u751f\u6210\u7684\u5de5\u5177\u8c03\u7528\u8ba1\u5212:
[
{
\"name\": \"search_google_news\",
\"arguments\": {\"keyword\": \"\u5c0f\u7c73SU7\"}
},
{
\"name\": \"analyze_sentiment\",
\"arguments\": {
\"text\": \"{{search_google_news}}\",
\"filename\": \"sentiment_\u5c0f\u7c73su7_20251113_213000.md\"
}
},
{
\"name\": \"send_email_with_attachment\",
\"arguments\": {
\"to\": \"user@qq.com\",
\"subject\": \"\u5c0f\u7c73SU7\u8fd1\u671f\u70ed\u70b9\u65b0\u95fb\u60c5\u611f\u5206\u6790\u62a5\u544a\",
\"body\": \"\u60a8\u597d\uff0c\u8fd9\u662f\u5206\u6790\u62a5\u544a\uff0c\u8bf7\u67e5\u6536\u3002\",
\"filename\": \"{{analyze_sentiment}}\"
}
}
]
\u200b
\u2705 \u5df2\u83b7\u53d6\u65b0\u95fb...
\u2705 \u5df2\u751f\u6210\u5206\u6790\u62a5\u544a...
\u2705 \u90ae\u4ef6\u5df2\u6210\u529f\u53d1\u9001...<\/pre>\n\n\n\n
\n\n\n\n\ud83c\udf93 \u6838\u5fc3\u77e5\u8bc6\u70b9\u603b\u7ed3<\/h2>\n\n\n\n
1. MCP Server \u8bbe\u8ba1\u8981\u70b9<\/h3>\n\n\n\n
\n
@mcp.tool()<\/code> \u88c5\u9970\u5668\u5b9a\u4e49\u5de5\u5177<\/li>\n\n\n\nmcp.run(transport='stdio')<\/code> \u542f\u52a8<\/li>\n<\/ul>\n\n\n\n2. MCP Client \u8bbe\u8ba1\u8981\u70b9<\/h3>\n\n\n\n
\n
StdioServerParameters<\/code> \u8fde\u63a5 Server<\/li>\n\n\n\nClientSession<\/code> \u7ba1\u7406\u4f1a\u8bdd<\/li>\n\n\n\nlist_tools()<\/code> \u83b7\u53d6\u53ef\u7528\u5de5\u5177<\/li>\n\n\n\n{{tool_name}}<\/code>\uff09<\/li>\n<\/ul>\n\n\n\n3. \u5de5\u5177\u8c03\u7528\u94fe\u89c4\u5212\u6280\u5de7<\/h3>\n\n\n\n
\n
4. \u5e38\u89c1\u95ee\u9898\u4e0e\u89e3\u51b3\u65b9\u6848<\/h3>\n\n\n\n
\u95ee\u9898<\/th> \u539f\u56e0<\/th> \u89e3\u51b3\u65b9\u6848<\/th><\/tr><\/thead> role='tool'<\/code> \u9519\u8bef<\/td>API \u8981\u6c42 tool \u6d88\u606f\u5fc5\u987b\u6709\u5bf9\u5e94 tool_calls<\/td> \u6539\u7528 role='assistant'<\/code> \u5e76\u6807\u6ce8\u5de5\u5177\u540d<\/td><\/tr>SMTP \u8fde\u63a5\u8d85\u65f6<\/td> \u670d\u52a1\u5668\u5730\u5740\u9519\u8bef<\/td> \u786e\u4fdd\u4f7f\u7528 smtp.qq.com<\/code> \u800c\u975e qq.com<\/code><\/td><\/tr>\u5de5\u5177\u672a\u88ab\u8c03\u7528<\/td> AI \u6ca1\u6709\u7406\u89e3\u4efb\u52a1<\/td> \u5f3a\u5316 system prompt\uff0c\u63d0\u4f9b\u793a\u4f8b<\/td><\/tr> \u6587\u4ef6\u8def\u5f84\u95ee\u9898<\/td> \u5de5\u5177\u95f4\u8def\u5f84\u683c\u5f0f\u4e0d\u4e00\u81f4<\/td> \u7edf\u4e00\u4f7f\u7528\u6587\u4ef6\u540d\uff08\u4e0d\u542b\u8def\u5f84\uff09<\/td><\/tr><\/tbody><\/table><\/figure>\n\n\n\n
\n\n\n\n\ud83d\udd25 \u8fdb\u9636\u6269\u5c55<\/h2>\n\n\n\n
1. \u6dfb\u52a0\u66f4\u591a\u5de5\u5177<\/h3>\n\n\n\n
@mcp.tool()
async def generate_chart(data: str) -> str:
\"\"\"\u6839\u636e\u6570\u636e\u751f\u6210\u56fe\u8868\"\"\"
# \u4f7f\u7528 matplotlib \u751f\u6210\u56fe\u8868
pass
\u200b
@mcp.tool()
async def translate_text(text: str, target_lang: str) -> str:
\"\"\"\u7ffb\u8bd1\u6587\u672c\"\"\"
# \u8c03\u7528\u7ffb\u8bd1 API
pass<\/pre>\n\n\n\n2. \u652f\u6301\u6d41\u5f0f\u54cd\u5e94<\/h3>\n\n\n\n
async def stream_process_query(self, query: str):
\"\"\"\u6d41\u5f0f\u5904\u7406\u67e5\u8be2\"\"\"
async for chunk in self.client.chat.completions.create(
model=self.model,
messages=messages,
stream=True
):
print(chunk.choices[0].delta.content, end=\"\")<\/pre>\n\n\n\n3. \u6dfb\u52a0\u9519\u8bef\u91cd\u8bd5\u673a\u5236<\/h3>\n\n\n\n
from tenacity import retry, stop_after_attempt, wait_exponential
\u200b
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def call_tool_with_retry(self, tool_name, tool_args):
\"\"\"\u5e26\u91cd\u8bd5\u7684\u5de5\u5177\u8c03\u7528\"\"\"
return await self.session.call_tool(tool_name, tool_args)<\/pre>\n\n\n\n4. \u5de5\u5177\u6743\u9650\u63a7\u5236<\/h3>\n\n\n\n
class ToolPermissionManager:
def __init__(self):
self.allowed_tools = {
\"user1\": [\"search_google_news\", \"analyze_sentiment\"],
\"admin\": [\"*\"] # \u6240\u6709\u5de5\u5177
}
def check_permission(self, user: str, tool_name: str) -> bool:
if \"*\" in self.allowed_tools.get(user, []):
return True
return tool_name in self.allowed_tools.get(user, [])<\/pre>\n\n\n\n
\n\n\n\n\ud83d\udcca \u6027\u80fd\u4f18\u5316\u5efa\u8bae<\/h2>\n\n\n\n
1. \u5e76\u884c\u6267\u884c\u72ec\u7acb\u5de5\u5177<\/h3>\n\n\n\n
import asyncio
\u200b
# \u5982\u679c\u5de5\u5177\u95f4\u65e0\u4f9d\u8d56\uff0c\u53ef\u5e76\u884c\u6267\u884c
results = await asyncio.gather(
self.session.call_tool(\"tool1\", args1),
self.session.call_tool(\"tool2\", args2)
)<\/pre>\n\n\n\n2. \u7f13\u5b58\u5de5\u5177\u7ed3\u679c<\/h3>\n\n\n\n
from functools import lru_cache
\u200b
@lru_cache(maxsize=100)
def get_cached_news(keyword: str):
# \u7f13\u5b58\u65b0\u95fb\u641c\u7d22\u7ed3\u679c
pass<\/pre>\n\n\n\n3. \u9650\u6d41\u63a7\u5236<\/h3>\n\n\n\n
from aiolimiter import AsyncLimiter
\u200b
limiter = AsyncLimiter(10, 60) # \u6bcf\u5206\u949f\u6700\u591a10\u6b21
\u200b
async def rate_limited_call(self, tool_name, args):
async with limiter:
return await self.session.call_tool(tool_name, args)<\/pre>\n\n\n\n
\n\n\n\n\ud83c\udfaf \u6700\u4f73\u5b9e\u8df5<\/h2>\n\n\n\n
\u2705 DO\uff08\u63a8\u8350\u505a\u6cd5\uff09<\/h3>\n\n\n\n
\n
search_google_news<\/code> \u800c\u975e tool1<\/code><\/li>\n\n\n\n\u274c DON’T\uff08\u907f\u514d\u505a\u6cd5\uff09<\/h3>\n\n\n\n
\n
\n\n\n\n\ud83e\uddea \u6d4b\u8bd5\u5efa\u8bae<\/h2>\n\n\n\n
\u5355\u5143\u6d4b\u8bd5\u793a\u4f8b<\/h3>\n\n\n\n
import pytest
from server import search_google_news
\u200b
@pytest.mark.asyncio
async def test_search_google_news():
result = await search_google_news(\"\u6d4b\u8bd5\u5173\u952e\u8bcd\")
assert \"\u2705\" in result or \"\u274c\" in result
\u200b
@pytest.mark.asyncio
async def test_analyze_sentiment():
text = \"\u8fd9\u662f\u4e00\u6761\u6b63\u9762\u65b0\u95fb\"
filename = \"test_report.md\"
result = await analyze_sentiment(text, filename)
assert result == filename<\/pre>\n\n\n\n\u96c6\u6210\u6d4b\u8bd5<\/h3>\n\n\n\n
@pytest.mark.asyncio
async def test_full_workflow():
client = MCPClient()
await client.connect_to_server(\"server.py\")
query = \"\u641c\u7d22\u5e76\u5206\u6790\u6d4b\u8bd5\u65b0\u95fb\"
response = await client.process_query(query)
assert \"\u2705\" in response
await client.cleanup()<\/pre>\n\n\n\n
\n\n\n\n\ud83d\udcda \u53c2\u8003\u8d44\u6e90<\/h2>\n\n\n\n
\n