from typing import Dict, Any, Optional, Union, List
from pydantic import BaseModel, Field, validator
from typing_extensions import Literal
class JSONRPCEnvelope(BaseModel):
"""JSON-RPC 2.0 envelope model"""
jsonrpc: Literal["2.0"] = "2.0"
id: Optional[Union[str, int]] = None
method: Optional[str] = None
params: Optional[Dict[str, Any]] = None
result: Optional[Any] = None
error: Optional[Dict[str, Any]] = None
@validator('method')
def method_required_for_request(cls, v, values):
# Method is required for requests (when result and error are None)
if v is None and values.get('result') is None and values.get('error') is None:
if values.get('id') is not None: # Not a notification
raise ValueError('method is required for requests')
return v
class InitializeRequest(BaseModel):
"""MCP initialize request parameters"""
protocolVersion: str = "2024-11-05"
capabilities: Dict[str, Any] = Field(default_factory=dict)
clientInfo: Dict[str, str] = Field(default_factory=dict)
class InitializeResponse(BaseModel):
"""MCP initialize response"""
protocolVersion: str = "2024-11-05"
capabilities: Dict[str, Any]
serverInfo: Dict[str, str]
class ResourceListRequest(BaseModel):
"""Request to list resources"""
cursor: Optional[str] = None
class ResourceReadRequest(BaseModel):
"""Request to read a resource"""
uri: str
class Resource(BaseModel):
"""Resource model"""
uri: str
name: str
description: Optional[str] = None
mimeType: Optional[str] = None
class ResourceContent(BaseModel):
"""Resource content model"""
uri: str
mimeType: str
text: str
class ToolCallRequest(BaseModel):
"""Tool call request"""
name: str
arguments: Dict[str, Any] = Field(default_factory=dict)
class ToolResult(BaseModel):
"""Tool execution result"""
content: List[Dict[str, Any]]
isError: bool = False
class CompletionRequest(BaseModel):
"""LLM completion request"""
prompt: str
max_tokens: Optional[int] = 1000
temperature: Optional[float] = 0.2
stream: bool = False
class CompletionChunk(BaseModel):
"""Streaming completion chunk"""
content: str
finished: bool = False