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

MCP实战-本地MCP Server + Client实战

news 2025/5/15 17:21:56

概述

本文开发一个MCP的Client和Server。然后通过本地模式来运行，并获取到server的结果。

MCP Server开发

import anyio
import click
import mcp.types as types
from mcp.server.lowlevel import Server
from pydantic import FileUrlSAMPLE_RESOURCES = {"greeting": "Hello! This is a sample text resource.","help": "This server provides a few sample text resources for testing.","about": "This is the simple-resource MCP server implementation.",
}@click.command()
@click.option("--port", default=10005, 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-simple-resource")@app.list_resources()async def list_resources() -> list[types.Resource]:return [types.Resource(uri=FileUrl(f"file:///var/domcp/{name}.txt"),name=name,description=f"A sample text resource named {name}",mimeType="text/plain",)for name in SAMPLE_RESOURCES.keys()]@app.read_resource()async def read_resource(uri: FileUrl) -> str | bytes:name = uri.path.replace(".txt", "").lstrip("/")if name not in SAMPLE_RESOURCES:raise ValueError(f"Unknown resource: {uri}")return SAMPLE_RESOURCES[name]if transport == "sse":from mcp.server.sse import SseServerTransportfrom starlette.applications import Starlettefrom 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__":main()

MCP Client开发

这里使用

import asyncio
from mcp.types import AnyUrl
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_clientasync def main():async with stdio_client(StdioServerParameters(command="python", args=["/root/domcp/server2.py"])) as (read, write):async with ClientSession(read, write) as session:await session.initialize()# List available resourcesresources = await session.list_resources()print(resources)print("----------------------")# Get a specific resourceresource = await session.read_resource(AnyUrl("file:///greeting.txt"))print(resource)asyncio.run(main())

运行MCP Client

运行Client后输出如下内容：

(ailab) root@tra:~/domcp# python client2.py 
meta=None nextCursor=None resources=[Resource(uri=Url('file:///root/domcp/greeting.txt'), name='greeting', description='A sample text resource named greeting', mimeType='text/plain', size=None, annotations=None), Resource(uri=Url('file:///root/domcp/help.txt'), name='help', description='A sample text resource named help', mimeType='text/plain', size=None, annotations=None), Resource(uri=Url('file:///root/domcp/about.txt'), name='about', description='A sample text resource named about', mimeType='text/plain', size=None, annotations=None)]----------------------
meta=None contents=[TextResourceContents(uri=Url('file:///greeting.txt'), mimeType='text/plain', text='Hello! This is a sample text resource.')]