MCP API¶
This section provides detailed API documentation for the Model Context Protocol (MCP) components in Scoras.
MCP Client¶
MCPClient¶
class MCPClient:
def __init__(
self,
server_url: str,
api_key: Optional[str] = None,
enable_scoring: bool = True
):
"""
Initialize an MCP client to connect to MCP servers.
Args:
server_url: URL of the MCP server
api_key: Optional API key for authentication
enable_scoring: Whether to track complexity scoring
"""
async def execute_tool(
self,
tool_name: str,
parameters: Dict[str, Any]
) -> Any:
"""
Execute a tool on the MCP server asynchronously.
Args:
tool_name: Name of the tool to execute
parameters: Parameters for the tool
Returns:
Result of the tool execution
"""
def execute_tool_sync(
self,
tool_name: str,
parameters: Dict[str, Any]
) -> Any:
"""
Execute a tool on the MCP server synchronously.
Args:
tool_name: Name of the tool to execute
parameters: Parameters for the tool
Returns:
Result of the tool execution
"""
async def stream_tool(
self,
tool_name: str,
parameters: Dict[str, Any]
) -> AsyncIterator[Any]:
"""
Stream the result of a tool execution asynchronously.
Args:
tool_name: Name of the tool to execute
parameters: Parameters for the tool
Yields:
Chunks of the tool execution result
"""
async def list_tools(self) -> List[Dict[str, Any]]:
"""
List available tools on the MCP server asynchronously.
Returns:
List of tool descriptions
"""
def list_tools_sync(self) -> List[Dict[str, Any]]:
"""
List available tools on the MCP server synchronously.
Returns:
List of tool descriptions
"""
async def get_server_info(self) -> Dict[str, Any]:
"""
Get information about the MCP server asynchronously.
Returns:
Dictionary containing server information
"""
def get_server_info_sync(self) -> Dict[str, Any]:
"""
Get information about the MCP server synchronously.
Returns:
Dictionary containing server information
"""
def get_complexity_score(self) -> Dict[str, Any]:
"""
Get the complexity score for the MCP client.
Returns:
Dictionary containing complexity score information
"""
MCP Server¶
MCPServer¶
class MCPServer:
def __init__(
self,
name: str,
description: str,
tools: List[Tool],
capabilities: Optional[List[str]] = None,
enable_scoring: bool = True
):
"""
Initialize an MCP server.
Args:
name: Name of the server
description: Description of the server
tools: List of tools to expose
capabilities: Optional list of capabilities ("tools", "streaming", "batching")
enable_scoring: Whether to track complexity scoring
"""
def add_tool(self, tool: Tool) -> None:
"""
Add a tool to the server.
Args:
tool: Tool to add
"""
async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""
Handle an MCP request asynchronously.
Args:
request: MCP request in JSON-RPC format
Returns:
MCP response in JSON-RPC format
"""
def get_complexity_score(self) -> Dict[str, Any]:
"""
Get the complexity score for the MCP server.
Returns:
Dictionary containing complexity score information
"""
MCPAgentServer¶
class MCPAgentServer(MCPServer):
def __init__(
self,
name: str,
description: str,
agent: Agent,
tools: Optional[List[Tool]] = None,
capabilities: Optional[List[str]] = None,
enable_scoring: bool = True
):
"""
Initialize an MCP server powered by an agent.
Args:
name: Name of the server
description: Description of the server
agent: Agent to power the server
tools: Optional list of additional tools to expose
capabilities: Optional list of capabilities ("tools", "streaming", "batching")
enable_scoring: Whether to track complexity scoring
"""
async def handle_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
"""
Handle a task request asynchronously.
Args:
task: Task request
Returns:
Task response
"""
MCP Functions¶
create_mcp_server¶
def create_mcp_server(
name: str,
description: str,
tools: List[Tool],
capabilities: Optional[List[str]] = None,
enable_scoring: bool = True
) -> MCPServer:
"""
Factory function to create an MCP server.
Args:
name: Name of the server
description: Description of the server
tools: List of tools to expose
capabilities: Optional list of capabilities ("tools", "streaming", "batching")
enable_scoring: Whether to track complexity scoring
Returns:
Instantiated MCPServer
"""
create_mcp_agent_server¶
def create_mcp_agent_server(
name: str,
description: str,
agent: Agent,
tools: Optional[List[Tool]] = None,
capabilities: Optional[List[str]] = None,
enable_scoring: bool = True
) -> MCPAgentServer:
"""
Factory function to create an MCP server powered by an agent.
Args:
name: Name of the server
description: Description of the server
agent: Agent to power the server
tools: Optional list of additional tools to expose
capabilities: Optional list of capabilities ("tools", "streaming", "batching")
enable_scoring: Whether to track complexity scoring
Returns:
Instantiated MCPAgentServer
"""
run_mcp_server¶
async def run_mcp_server(
server: Union[MCPServer, MCPAgentServer],
host: str = "0.0.0.0",
port: int = 8000
) -> None:
"""
Run an MCP server asynchronously.
Args:
server: MCP server to run
host: Host to bind to
port: Port to listen on
"""