"""
Standard response envelope for all MCP tools.
This provides a consistent structure for tool responses, making it easier
for the LLM to parse results and for debugging/monitoring.
Benefits:
- Consistent response format across all tools
- Built-in metadata for tracing and debugging
- Clear error handling with missing_reason field
- Presentation layer for human-readable output
"""
from __future__ import annotations
from typing import Any, Dict, Generic, Optional, TypeVar
from datetime import datetime, timezone
from pydantic import BaseModel, Field
T = TypeVar("T")
class Meta(BaseModel):
"""Metadata about the tool execution."""
schema_version: str = Field(
default="1.0.0",
description="Envelope schema version"
)
tool_name: str = Field(
description="Name of the tool that generated this response"
)
trace_id: str = Field(
description="Unique identifier for tracing this request"
)
generated_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
description="When this response was generated"
)
source: str = Field(
default="updation_mcp",
description="Source system identifier"
)
execution_time_ms: Optional[int] = Field(
default=None,
description="Tool execution time in milliseconds"
)
class Envelope(BaseModel, Generic[T]):
"""
Standard envelope for all tool responses.
This wraps the actual data with metadata and optional error information.
The generic type T allows for type-safe data payloads.
Example:
```python
# Success case
envelope = Envelope(
meta=Meta(tool_name="get_user", trace_id="abc-123"),
data={"user_id": 123, "name": "John"},
missing_reason=None
)
# Error case
envelope = Envelope(
meta=Meta(tool_name="get_user", trace_id="abc-123"),
data={},
missing_reason="User not found"
)
```
"""
meta: Meta = Field(
description="Metadata about this response"
)
data: T = Field(
description="The actual response data"
)
missing_reason: Optional[str] = Field(
default=None,
description="If data is missing/incomplete, explain why"
)
@property
def is_success(self) -> bool:
"""Check if this response represents a successful operation."""
return self.missing_reason is None
@property
def is_error(self) -> bool:
"""Check if this response represents an error."""
return self.missing_reason is not None
def wrap_response(
*,
tool_name: str,
trace_id: str,
data: Any,
missing_reason: Optional[str] = None,
execution_time_ms: Optional[int] = None,
) -> Dict[str, Any]:
"""
Convenience function to create an envelope and return as dict.
This is the recommended way to return data from MCP tools.
Args:
tool_name: Name of the tool generating the response
trace_id: Unique trace ID for this request
data: The actual response data (will be serialized)
missing_reason: Optional error/missing data explanation
execution_time_ms: Optional execution time tracking
Returns:
dict: Envelope as a plain dictionary for JSON serialization
Example:
```python
@mcp.tool()
async def get_user(user_id: int, trace_id: str):
user = await fetch_user(user_id)
return wrap_response(
tool_name="get_user",
trace_id=trace_id,
data=user,
)
```
"""
meta = Meta(
tool_name=tool_name,
trace_id=trace_id,
execution_time_ms=execution_time_ms,
)
envelope = Envelope(
meta=meta,
data=data,
missing_reason=missing_reason,
)
return envelope.model_dump()
