-
Notifications
You must be signed in to change notification settings - Fork 2.2k
Open
Description
Description
Some mcp tool, i don't want let LLM know, So i want transfer data by metadata, How can it do it? thank you !
Server example:
import datetime
from fastmcp import FastMCP, Context
from mcp.types import CallToolResult
mcp = FastMCP("EchoServer")
@mcp.tool()
async def echo_tool(text: str, ctx: Context) -> CallToolResult:
# 从 ctx 获取 metadata
meta_in = ctx.request_meta or {}
print("[Server] meta_in:", meta_in)
return CallToolResult(
content=[{"type": "text", "text": text.upper()}],
meta={
"user_id": meta_in.get("user_id", "unknown"),
"server_timestamp": datetime.datetime.utcnow().isoformat() + "Z"
}
)
if __name__ == "__main__":
mcp.run(transport="sse")
Client Example
# client.py
from mcp.types import CallToolRequest, CallToolRequestParams, RequestParams
import asyncio
from fastmcp import Client
from mcp.types import CallToolRequestParams, RequestParams
import json
def simulate():
# 构造调用请求,带上 meta 信息
req = CallToolRequest(
method="tools/call",
params=CallToolRequestParams(
name="echo_tool",
arguments={"text": "hello MCP"},
meta=RequestParams.Meta( # 通过 alias="_meta" 被序列化为 "_meta"
user_id="alice",
session_id="sess-123",
),
),
)
# 序列化为 JSON-RPC 并发送
payload = req.model_dump(by_alias=True)
print("Sending:", json.dumps(payload, indent=2))
# 这里你可以用 WebSocket/httpx/aiohttp 发送 payload 到 MCP server
# client_sse.py
async def main():
# 指定 SSE 地址(在 server.run(transport="sse") 中默认挂载为 http://localhost:8000/sse)
async with Client("http://localhost:8000/sse") as client:
# 调用 echo_tool,自动使用 SSE transport
result = await client.call_tool(
"echo_tool",
{"text": "hello MCP"},
meta=RequestParams.Meta(
user_id="alice",
session_id="sess-123"
)
)
print("Response content:", result.content)
print("Response meta:", result.meta)
if __name__ == "__main__":
asyncio.run(main())
References
No response
Metadata
Metadata
Assignees
Labels
No labels