当前位置： 首页 > wzjs >正文

湖南城乡和建设厅网站校园网站开发技术

wzjs 2025/9/9 22:40:08
湖南城乡和建设厅网站,校园网站开发技术,网站建设有什么理论依据,淘宝客网站整站源码引言 最近MCP大火，本文尝试揭开它神秘的面纱。文章较长，分为上下两篇。这是第二篇。 MCP实战 MCP有通过高级API和底层API实现两种方法，我们先来看下底层API如何实现。 底层 API实现 服务器端： server.py: import anyio # …

引言

最近MCP大火，本文尝试揭开它神秘的面纱。文章较长，分为上下两篇。这是第二篇。

MCP实战

MCP有通过高级API和底层API实现两种方法，我们先来看下底层API如何实现。

底层 API实现

服务器端：

server.py:

import anyio  # AnyIO is an asynchronous networking and concurrency library that works on top of either asyncio or trio.
import click  # Click is a Python package for creating beautiful command line interfaces in a composable way with as little code as necessary.
import httpx
import mcp.types as types
from mcp.server.lowlevel import Serverfrom datetime import datetime
from tavily import TavilyClient
import os
import jsonfrom dotenv import load_dotenvload_dotenv()tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))def get_now() -> list[types.TextContent]:return_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")return [types.TextContent(type="text", text=return_str)]def web_search(query: str) -> list[types.TextContent]:response = tavily_client.search(query)results = response.get("results")return [types.TextContent(type="text", text=result.get("content")) for result in results]@click.command()
@click.option("--port", default=8000, help="Port to listen on for SSE")
@click.option("--transport",type=click.Choice(["stdio", "sse"]),default="stdio",help="Transport type",
)
def main(port: int, transport: str) -> int:app = Server("mcp-test-server")@app.call_tool()async def fetch_tool(name: str, arguments: dict) -> list[types.TextContent]:if name == "get_now":return get_now()elif name == "web_search":return web_search(arguments["query"])else:raise ValueError(f"Unkonw tool: {name}")@app.list_tools()async def list_tools() -> list[types.Tool]:return [types.Tool(name="web_search",description="进行谷歌搜索，可以查询最近发生的实事、天气等",inputSchema={"type": "object","required": ["query"],"properties": {"query": {"type": "string","description": "要进行互联网搜索的查询",}},},),types.Tool(name="get_now",description="获取当前时间",inputSchema={},),]if transport == "sse":from mcp.server.sse import SseServerTransportfrom starlette.applications import (Starlette,)  # Starlette is a lightweight ASGI framework/toolkit, which is ideal for building async web services in Python.from starlette.routing import Mount, Routesse = SseServerTransport("/messages/")async def handle_sse(request):async with sse.connect_sse(request.scope, request.receive, request._send) as streams:await app.run(streams[0], streams[1], app.create_initialization_options())starlette_app = Starlette(debug=True,routes=[Route("/sse", endpoint=handle_sse),Mount("/messages/", app=sse.handle_post_message),],)import uvicornuvicorn.run(starlette_app, host="0.0.0.0", port=port)else:from mcp.server.stdio import stdio_serverasync def arun():async with stdio_server() as streams:await app.run(streams[0], streams[1], app.create_initialization_options())anyio.run(arun)return 0if __name__ == "__main__":import syssys.exit(main())# python -m server --transport sse

服务器端支持sse和stdio，如果以sse启动： python -m server --transport sse。

客户端：

client.py:

import asyncio
import json
import os
import sys
import logging
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession
from mcp.client.sse import sse_client
from openai import AsyncOpenAI
from dotenv import load_dotenvload_dotenv()# 配置日志系统
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)class MCPChatClient:def __init__(self):self.session: Optional[ClientSession] = Noneself.exit_stack = AsyncExitStack()# 初始化 OpenAI 客户端self.openai = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"),base_url=os.getenv("OPENAI_BASE_URL"),)async def __aenter__(self):await self.exit_stack.__aenter__()return selfasync def __aexit__(self, exc_type, exc_val, exc_tb):await self.exit_stack.__aexit__(exc_type, exc_val, exc_tb)async def connect(self, server_url: str):logger.info(f"Connecting to SSE server at {server_url}...")# 连接 SSE 服务端并创建 MCP 会话streams = await self.exit_stack.enter_async_context(sse_client(url=server_url))self.session = await self.exit_stack.enter_async_context(ClientSession(*streams))await self.session.initialize()# 获取并打印可用工具列表response = await self.session.list_tools()tools = response.toolslogger.info(f"Connected to server with tools: {[tool.name for tool in tools]}")async def handle_query(self, user_input: str) -> str:messages = [{"role": "user", "content": user_input}]response = await self.session.list_tools()# 将 MCP 工具列表格式化为 OpenAI function calling 格式tools_payload = [{"type": "function","function": {"name": tool.name,"description": tool.description,"parameters": tool.inputSchema,},}for tool in response.tools]logger.debug(f"Available tools: {json.dumps(tools_payload, indent=2)}")# 首次调用 OpenAI Chat Completionchat_response = await self.openai.chat.completions.create(model=os.getenv("OPENAI_MODEL"),max_tokens=1000,messages=messages,tools=tools_payload,)output_texts = []assistant_msg = chat_response.choices[0].message# 检查是否触发了工具调用if assistant_msg.tool_calls:for tool_call in assistant_msg.tool_calls:tool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)try:# 执行工具调用result = await self.session.call_tool(tool_name, tool_args)output_texts.append(f"[Called {tool_name} with args {tool_args}]")# 将工具调用响应添加到对话中messages.extend([{"role": "assistant","content": None,"tool_calls": [tool_call],},{"role": "tool","tool_call_id": tool_call.id,"content": result.content[0].text,},])logger.info(f"Tool {tool_name} returned: {result.content[0].text}")# 根据工具响应再次请求 OpenAI，继续对话chat_response = await self.openai.chat.completions.create(model=os.getenv("OPENAI_MODEL"),max_tokens=1000,messages=messages,)content = chat_response.choices[0].message.contentoutput_texts.append(str(content))except Exception as e:logger.exception(f"Error calling tool {tool_name} with args {tool_args}.")else:# 没有工具调用，直接返回 Assistant 的响应content = assistant_msg.contentoutput_texts.append(str(content))return "\n".join(output_texts)async def interactive_chat(self):print("\nMCP Chat Client Started!")print("Type your queries or 'q' to exit.")while True:try:query = input("\nQuery: ").strip()if query.lower() == "q":breakresponse = await self.handle_query(query)print("\n" + response)except Exception as e:logger.exception("Unexpected error during chat interaction.")async def main():if len(sys.argv) < 2:print("Usage: python -m client <SSE MCP server URL>")sys.exit(1)try:async with MCPChatClient() as client:await client.connect(server_url=sys.argv[1])await client.interactive_chat()except Exception:logger.exception("Failed to start MCPChatClient.")if __name__ == "__main__":asyncio.run(main())

假设服务器以sse启动，通过python -m client http://localhost:8000/sse启动客户端。这里示例了通过OpenAI协议中的函数调用方式来实现MCP的工具调用，实际上还可通过ReACT等方式。

fastmcp实现

MCP的sdk提供了高级API实现，可以快速编写服务器：

fast_server.py:

from mcp.server.fastmcp import FastMCPfrom datetime import datetime
from tavily import TavilyClient
import osfrom dotenv import load_dotenvload_dotenv()tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))mcp = FastMCP("test-demo", port="8088")@mcp.tool()
def get_now() -> str:"""获取当前时间Returns:str: %Y-%m-%d %H:%M:%S 格式的时间"""return datetime.now().strftime("%Y-%m-%d %H:%M:%S")@mcp.tool()
def web_search(query: str) -> list[str]:"""进行谷歌搜索，可以查询最近发生的实事、天气等Args:query (str): 要进行互联网搜索的查询Returns:list[str]: 查询结果列表"""response = tavily_client.search(query)results = response.get("results")return [result.get("content") for result in results]if __name__ == "__main__":# Initialize and run the servermcp.run(transport="sse")

可以看到这里我们主要关心的就是如何定义好工具。

首先通过python fast_server.py启动服务端，然后通过python -m client http://localhost:8088/sse启动客户端。

客户端日志：

> python -m client http://localhost:8088/sse
INFO:__main__:Connecting to SSE server at http://localhost:8088/sse...
INFO:mcp.client.sse:Connecting to SSE endpoint: http://localhost:8088/sse
INFO:httpx:HTTP Request: GET http://localhost:8088/sse "HTTP/1.1 200 OK"
INFO:mcp.client.sse:Received endpoint URL: http://localhost:8088/messages/?session_id=6f7720204b1b4e6ab53ccd65d7a4c3a7
INFO:mcp.client.sse:Starting post writer with endpoint URL: http://localhost:8088/messages/?session_id=6f7720204b1b4e6ab53ccd65d7a4c3a7
INFO:httpx:HTTP Request: POST http://localhost:8088/messages/?session_id=6f7720204b1b4e6ab53ccd65d7a4c3a7 "HTTP/1.1 202 Accepted"
INFO:httpx:HTTP Request: POST http://localhost:8088/messages/?session_id=6f7720204b1b4e6ab53ccd65d7a4c3a7 "HTTP/1.1 202 Accepted"
INFO:httpx:HTTP Request: POST http://localhost:8088/messages/?session_id=6f7720204b1b4e6ab53ccd65d7a4c3a7 "HTTP/1.1 202 Accepted"
INFO:__main__:Connected to server with tools: ['get_now', 'web_search']

服务端日志：

> python fast_server.py
INFO:     Started server process [97933]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8088 (Press CTRL+C to quit)
INFO:     127.0.0.1:50789 - "GET /sse HTTP/1.1" 200 OK
INFO:     127.0.0.1:50791 - "POST /messages/?session_id=6f7720204b1b4e6ab53ccd65d7a4c3a7 HTTP/1.1" 202 Accepted
INFO:     127.0.0.1:50793 - "POST /messages/?session_id=6f7720204b1b4e6ab53ccd65d7a4c3a7 HTTP/1.1" 202 Accepted
INFO:     127.0.0.1:50795 - "POST /messages/?session_id=6f7720204b1b4e6ab53ccd65d7a4c3a7 HTTP/1.1" 202 Accepted
[04/08/25 15:32:04] INFO     Processing request of type ListToolsRequest

从上面的日志可以看到：

  1. 客户端通过http://localhost:8088/sse建立连接
  2. 服务端返回带session_id的URL： http://localhost:8088/messages/?session_id=6f7720204b1b4e6ab53ccd65d7a4c3a7
  3. 客户端通过这个端点发送POST请求，进入初始化阶段(能力协商等)。
  4. 然后发送了ListToolsRequest请求获取工具列表。
    • 这里返回了服务端定义的两个工具

然后假设用户输入了一个问题(客户端日志)：

MCP Chat Client Started!
Type your queries or 'q' to exit.Query: 现在几点了
INFO:httpx:HTTP Request: POST http://localhost:8088/messages/?session_id=5fd44bdc7c564f4c9feea03e59b6fc88 "HTTP/1.1 202 Accepted"
INFO:httpx:HTTP Request: POST http://***/v1/chat/completions "HTTP/1.1 200 "
INFO:httpx:HTTP Request: POST http://localhost:8088/messages/?session_id=5fd44bdc7c564f4c9feea03e59b6fc88 "HTTP/1.1 202 Accepted"
INFO:__main__:Tool get_now returned: 2025-04-08 16:11:54
INFO:httpx:HTTP Request: POST http://***/v1/chat/completions "HTTP/1.1 200 "[Called get_now with args {}]
现在的时间是2025年4月8日16点11分54秒。 Query:

服务端日志：

INFO:     127.0.0.1:60904 - "POST /messages/?session_id=5fd44bdc7c564f4c9feea03e59b6fc88 HTTP/1.1" 202 Accepted
[04/08/25 16:11:51] INFO     Processing request of type ListToolsRequest                                                                                                       server.py:534
INFO:     127.0.0.1:60925 - "POST /messages/?session_id=5fd44bdc7c564f4c9feea03e59b6fc88 HTTP/1.1" 202 Accepted
[04/08/25 16:11:54] INFO     Processing request of type CallToolRequest

这里用户输入了一个问题，实际上执行过程如下：

  1. 通过带session_id的URL发送POST请求获取工具列表(ListToolsRequest)
  2. 服务端返回工具列表
  3. 调用LLM来决定是否需要调用工具
  4. (这里需要调用工具)发送CallToolRequest
  5. 服务端处理CallToolRequest，执行工具调用并返回结果
  6. 客户端对工具调用结果进行渲染，返回给用户

参考

  1. https://modelcontextprotocol.io/
  2. https://spec.modelcontextprotocol.io/specification/2025-03-26/
  3. https://github.com/sidharthrajaram/mcp-sse

文章转载自：

http://i40BdAv4.qphdp.cn
http://LLweA84d.qphdp.cn
http://xs2WagHu.qphdp.cn
http://OtKbgNOj.qphdp.cn
http://0Kb14Xep.qphdp.cn
http://xY5G9B9k.qphdp.cn
http://TgvmJCQ0.qphdp.cn
http://pU8Cq9oy.qphdp.cn
http://sBSShD3u.qphdp.cn
http://iqZyXQuT.qphdp.cn
http://8qvuTcJl.qphdp.cn
http://CfCUAsUc.qphdp.cn
http://OBf7sDhF.qphdp.cn
http://vvPKlgyQ.qphdp.cn
http://JZXGpN88.qphdp.cn
http://xuy0U41r.qphdp.cn
http://xh87uEBh.qphdp.cn
http://PVhRuIfP.qphdp.cn
http://4zZxNHXt.qphdp.cn
http://UdJOpvNp.qphdp.cn
http://2bu4MH4H.qphdp.cn
http://rSGHaJFI.qphdp.cn
http://U1LRdEZM.qphdp.cn
http://MFX1fyws.qphdp.cn
http://NcJNPP1n.qphdp.cn
http://JKSm4rmg.qphdp.cn
http://lloukXW6.qphdp.cn
http://paSLIyuf.qphdp.cn
http://BbyRiLRw.qphdp.cn
http://nxLUopAx.qphdp.cn
http://www.dtcms.com/wzjs/672999.html

相关文章：

  • 网站主关键词网站建设案例哪家好
  • 个人网站设计文字内容模板做行业分析的网站
  • 绿色大气漂亮dedecms茶叶企业网站租二级目录做网站
  • 建立网站用主机宝建立的网站上传之后404
  • 徐州建设工程网站百度网盘怎么找资源
  • 网站上有什么作用二级网站建设检查评比方案
  • 戴尔的网站建设多用户商城系统的服务态度
  • 大品牌网站建设保险网站有哪些
  • 做网站最好的工具宾爵手表官方网站
  • 南京制作网站建站模板公司重庆seo杨洋
  • asp网站安装教程wordpress使postid顺序
  • 昆山品牌网站中国营销传播网官网
  • 做视觉影像网站用什么软件系统正国级领导有几位
  • dede网站图标wordpress登录链接修改
  • 大学文明校园网站建设方案网络营销是一种什么专业
  • 网站系统中备案申请表做的网站一模一样会被告吗
  • 网页设计网站含义网页网站的制作过程
  • 免费网站制作软件的app模板网站可以做seo吗
  • 深圳平台网站建设外包网站配置域名
  • 宿州科技网站建设公司国际网站怎么做
  • 黄山建设网站公司电话号码机械设备上海网站建设
  • 公司建设网站的服务费广州番禺建网站
  • 南充网站建设与维护优秀网站案例欣赏
  • 电商网站模板下载东莞网站包年优化
  • 做外贸网站公司哪家消费全返的 微网站开发
  • 韶关网站seo高校工会网站建设
  • 建设京东物流网站的目标是什么荣成网站开发
  • ps免费模板网站网上花店网页制作素材
  • 郑州做网站企起wordpress合并压缩
  • 怎么做网站账号注册机网站建设的博客