mirror of
https://github.com/mcp-use/mcp-use.git
synced 2026-09-14 18:06:01 +08:00
Pietro Zullo
ebef017114
* chore: add example server pretty print JSON RPC for transparency * updated telemetry module * chore: agent side tel - langchain adapter - client tel * server side tel: * update API reference * python transports and primitive reports are publishes only if not skipped * remove wrongly generate files * properly typed middleware * remove examples * feat(docs): add middleware and session management documentation - Introduced new documentation for middleware, detailing hooks, minimal examples, and best practices for integrating middleware in MCP servers. - Added a session management guide explaining session ID handling, usage patterns, and examples for managing per-client state in MCP servers. - Updated to include links to the new middleware and session management pages. * generate docs * remove few events * adapter tel * adapter tel * api ref docs update * google adapter fix * fix version * fix event naming * server example is streamable and echo tool fix * removed initialize middleware hook, initialize requests are not passed thorugh normal handlers * api ref * beta on middleware
73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from typing import Any
|
|
|
|
from mcp_use.server.middleware.middleware import CallNext, Middleware, ServerMiddlewareContext
|
|
from mcp_use.telemetry.telemetry import Telemetry
|
|
|
|
_telemetry = Telemetry()
|
|
|
|
|
|
class TelemetryMiddleware(Middleware):
|
|
async def on_call_tool(self, context: ServerMiddlewareContext[Any], call_next: CallNext[Any, Any]) -> Any:
|
|
start_time = time.time()
|
|
success = True
|
|
error_type: str | None = None
|
|
|
|
try:
|
|
return await call_next(context)
|
|
except Exception as exc:
|
|
success = False
|
|
error_type = type(exc).__name__
|
|
raise
|
|
finally:
|
|
tool_name = context.message.name
|
|
arguments = context.message.arguments or {}
|
|
serialized_arguments = json.dumps(arguments)
|
|
_telemetry.track_server_tool_call(
|
|
tool_name=tool_name,
|
|
length_input_argument=len(serialized_arguments),
|
|
success=success,
|
|
execution_time_ms=int((time.time() - start_time) * 1000),
|
|
error_type=error_type,
|
|
)
|
|
|
|
async def on_read_resource(self, context: ServerMiddlewareContext[Any], call_next: CallNext[Any, Any]) -> Any:
|
|
success = True
|
|
error_type: str | None = None
|
|
|
|
try:
|
|
return await call_next(context)
|
|
except Exception as exc:
|
|
success = False
|
|
error_type = type(exc).__name__
|
|
raise
|
|
finally:
|
|
_telemetry.track_server_resource_call(
|
|
name=str(context.message.uri),
|
|
description=None,
|
|
contents=[],
|
|
success=success,
|
|
error_type=error_type,
|
|
)
|
|
|
|
async def on_get_prompt(self, context: ServerMiddlewareContext[Any], call_next: CallNext[Any, Any]) -> Any:
|
|
success = True
|
|
error_type: str | None = None
|
|
|
|
try:
|
|
return await call_next(context)
|
|
except Exception as exc:
|
|
success = False
|
|
error_type = type(exc).__name__
|
|
raise
|
|
finally:
|
|
_telemetry.track_server_prompt_call(
|
|
name=context.message.name,
|
|
description=None,
|
|
success=success,
|
|
error_type=error_type,
|
|
)
|