Skip to content

MCP Examples

This page provides detailed examples of using the Model Context Protocol (MCP) with Scoras. These examples demonstrate how to create MCP servers, connect to them with clients, and integrate MCP with Scoras agents.

Basic MCP Server Example

This example shows how to create a simple MCP server with tools:

import asyncio
import json
from typing import Dict, Any, List

import scoras as sc
from scoras.mcp import create_mcp_server, run_mcp_server

# Define tools for our MCP server
@sc.tool(name="calculator", description="Perform basic arithmetic operations", complexity="simple")
async def calculator(operation: str, a: float, b: float) -> float:
    """
    Perform basic arithmetic operations.

    Args:
        operation: Operation to perform (add, subtract, multiply, divide)
        a: First number
        b: Second number

    Returns:
        Result of the operation
    """
    if operation == "add":
        return a + b
    elif operation == "subtract":
        return a - b
    elif operation == "multiply":
        return a * b
    elif operation == "divide":
        if b == 0:
            raise ValueError("Cannot divide by zero")
        return a / b
    else:
        raise ValueError(f"Unknown operation: {operation}")

@sc.tool(name="weather", description="Get weather information for a location", complexity="standard")
async def weather(location: str) -> Dict[str, Any]:
    """
    Get weather information for a location.

    Args:
        location: Location to get weather for

    Returns:
        Weather information
    """
    # In a real implementation, this would call a weather API
    # For this example, we'll return mock data
    return {
        "location": location,
        "temperature": 72,
        "conditions": "Sunny",
        "humidity": 45,
        "wind_speed": 5
    }

async def main():
    # Create an MCP server
    server = create_mcp_server(
        name="ScorasServer",
        description="Scoras MCP server with various tools",
        tools=[calculator, weather],
        capabilities=["tools", "streaming"],
        enable_scoring=True
    )

    # Print the server info
    print("Server Info:", server.get_server_info())

    # Print available tools
    print("Available Tools:", json.dumps(server.get_available_tools(), indent=2))

    # Execute a tool directly on the server
    print("\nExecuting calculator tool directly on server...")
    result = await server.execute_tool(
        tool_name="calculator",
        parameters={
            "operation": "multiply",
            "a": 5,
            "b": 7
        }
    )
    print("Result:", result)

    # Get the complexity score
    score = server.get_complexity_score()
    print("\nServer Complexity Score:", json.dumps(score, indent=2))

    # Run the server
    print("\nStarting MCP server on http://0.0.0.0:8000")
    await run_mcp_server(server, host="0.0.0.0", port=8000)

if __name__ == "__main__":
    asyncio.run(main())

MCP Client Example

This example shows how to connect to an MCP server and use its tools:

import asyncio
import json
from typing import Dict, Any

import scoras as sc
from scoras.mcp import MCPClient

async def main():
    # Create an MCP client
    client = MCPClient(
        server_url="http://localhost:8000",
        enable_scoring=True
    )

    # Get server info
    server_info = await client.get_server_info()
    print("Server Info:", server_info)

    # Get available tools
    tools = await client.get_available_tools()
    print("Available Tools:", json.dumps(tools, indent=2))

    # Execute calculator tool
    print("\nExecuting calculator tool...")
    calc_result = await client.execute_tool(
        tool_name="calculator",
        parameters={
            "operation": "multiply",
            "a": 5,
            "b": 7
        }
    )
    print("Result:", calc_result)

    # Execute weather tool
    print("\nExecuting weather tool...")
    weather_result = await client.execute_tool(
        tool_name="weather",
        parameters={
            "location": "New York"
        }
    )
    print("Result:", weather_result)

    # Get the complexity score
    score = client.get_complexity_score()
    print("\nClient Complexity Score:", json.dumps(score, indent=2))

if __name__ == "__main__":
    asyncio.run(main())

MCP Context Management Example

This example demonstrates how to use MCP context for maintaining conversation state:

import asyncio
import json

import scoras as sc
from scoras.mcp import MCPClient, MCPContext

async def main():
    # Create an MCP client
    client = MCPClient(
        server_url="http://localhost:8000",
        enable_scoring=True
    )

    # Create an MCP context
    context = MCPContext()

    # Add messages to the context
    context.add_user_message("I need to perform some calculations.")
    context.add_assistant_message("I can help with calculations. What would you like to calculate?")
    context.add_user_message("What is 5 multiplied by 7?")

    # Execute calculator tool with context
    print("\nExecuting calculator tool with context...")
    result = await client.execute_tool_with_context(
        context=context,
        tool_name="calculator",
        parameters={
            "operation": "multiply",
            "a": 5,
            "b": 7
        }
    )
    print("Result:", result)

    # Add the tool result to the context
    context.add_tool_result(result)

    # Add more messages
    context.add_assistant_message(f"The result of 5 multiplied by 7 is {result['result']}.")
    context.add_user_message("Now divide this result by 3.")

    # Execute another calculation with the updated context
    print("\nExecuting another calculation with updated context...")
    result2 = await client.execute_tool_with_context(
        context=context,
        tool_name="calculator",
        parameters={
            "operation": "divide",
            "a": result['result'],
            "b": 3
        }
    )
    print("Result:", result2)

    # Print the full context
    print("\nFull Context:")
    for message in context.messages:
        print(f"{message['role']}: {message.get('content', '')} {message.get('function_call', '')}")

if __name__ == "__main__":
    asyncio.run(main())

MCP Agent Adapter Example

This example shows how to adapt a Scoras agent to use MCP tools:

import asyncio
import json

import scoras as sc
from scoras.mcp import MCPAgentAdapter

async def main():
    # Create a Scoras agent
    agent = sc.Agent(
        model="openai:gpt-4o",
        system_prompt="You are a helpful assistant that can perform calculations and check weather.",
        enable_scoring=True
    )

    # Create an MCP agent adapter
    adapter = MCPAgentAdapter(
        agent=agent,
        enable_scoring=True
    )

    # Connect to an MCP server
    print("Connecting to MCP server...")
    await adapter.connect_to_server("http://localhost:8000")

    # List available tools
    tools = adapter.get_available_tools()
    print("Available Tools:", json.dumps(tools, indent=2))

    # Run the agent with a query that will use the calculator tool
    print("\nRunning agent with calculator query...")
    calc_response = await adapter.run("What is 5 multiplied by 7?")
    print("Response:", calc_response)

    # Run the agent with a query that will use the weather tool
    print("\nRunning agent with weather query...")
    weather_response = await adapter.run("What's the weather like in New York?")
    print("Response:", weather_response)

    # Get the complexity score
    score = adapter.get_complexity_score()
    print("\nAdapter Complexity Score:", json.dumps(score, indent=2))

if __name__ == "__main__":
    asyncio.run(main())

Streaming MCP Example

This example demonstrates streaming responses from an MCP server:

import asyncio
import json

import scoras as sc
from scoras.mcp import MCPClient

async def main():
    # Create an MCP client
    client = MCPClient(
        server_url="http://localhost:8000",
        enable_scoring=True
    )

    # Stream a response from the server
    print("\nStreaming response from server...")
    async for chunk in client.stream_tool(
        tool_name="calculator",
        parameters={
            "operation": "multiply",
            "a": 5,
            "b": 7
        }
    ):
        print("Chunk:", chunk)

    # Stream with context
    from scoras.mcp import MCPContext
    context = MCPContext()
    context.add_user_message("I need to calculate something.")

    print("\nStreaming with context...")
    async for chunk in client.stream_tool_with_context(
        context=context,
        tool_name="calculator",
        parameters={
            "operation": "multiply",
            "a": 5,
            "b": 7
        }
    ):
        print("Chunk:", chunk)

if __name__ == "__main__":
    asyncio.run(main())

Advanced MCP Server Example

This example shows a more advanced MCP server with custom handlers:

import asyncio
import json
from typing import Dict, Any, List, Optional

import scoras as sc
from scoras.mcp import create_mcp_server, run_mcp_server, MCPContext

# Define a custom handler for the server
class CustomMCPHandler:
    def __init__(self):
        self.request_count = 0

    async def pre_request_hook(self, method: str, params: Dict[str, Any]) -> None:
        """Called before handling a request."""
        self.request_count += 1
        print(f"Request #{self.request_count}: {method}")

    async def post_request_hook(self, method: str, params: Dict[str, Any], result: Any) -> None:
        """Called after handling a request."""
        print(f"Completed request: {method}")

    async def error_hook(self, method: str, params: Dict[str, Any], error: Exception) -> Optional[Dict[str, Any]]:
        """Called when an error occurs."""
        print(f"Error in {method}: {str(error)}")
        # Return a custom error response or None to use the default
        return {
            "error": {
                "code": 500,
                "message": f"Custom error handler: {str(error)}"
            }
        }

# Define tools
@sc.tool(name="calculator", description="Perform calculations", complexity="simple")
async def calculator(operation: str, a: float, b: float) -> float:
    if operation == "add":
        return a + b
    elif operation == "subtract":
        return a - b
    elif operation == "multiply":
        return a * b
    elif operation == "divide":
        if b == 0:
            raise ValueError("Cannot divide by zero")
        return a / b
    else:
        raise ValueError(f"Unknown operation: {operation}")

async def main():
    # Create a custom handler
    handler = CustomMCPHandler()

    # Create an MCP server with the custom handler
    server = create_mcp_server(
        name="AdvancedScorasServer",
        description="Advanced Scoras MCP server with custom handler",
        tools=[calculator],
        capabilities=["tools", "streaming"],
        enable_scoring=True,
        pre_request_hook=handler.pre_request_hook,
        post_request_hook=handler.post_request_hook,
        error_hook=handler.error_hook
    )

    # Run the server
    print("Starting advanced MCP server on http://0.0.0.0:8000")
    await run_mcp_server(server, host="0.0.0.0", port=8000)

if __name__ == "__main__":
    asyncio.run(main())

MCP with Workflow Example

This example demonstrates integrating MCP with Scoras workflows:

import asyncio
import json
from pydantic import BaseModel

import scoras as sc
from scoras.mcp import create_mcp_server, run_mcp_server

# Define a state model for our workflow
class CalculationState(BaseModel):
    operation: str
    a: float
    b: float
    result: float = 0.0

# Define workflow nodes
async def validate_input(state):
    """Validate the input parameters."""
    if state.operation not in ["add", "subtract", "multiply", "divide"]:
        raise ValueError(f"Invalid operation: {state.operation}")
    if state.operation == "divide" and state.b == 0:
        raise ValueError("Cannot divide by zero")
    return {}

async def perform_calculation(state):
    """Perform the calculation."""
    if state.operation == "add":
        result = state.a + state.b
    elif state.operation == "subtract":
        result = state.a - state.b
    elif state.operation == "multiply":
        result = state.a * state.b
    else:  # divide
        result = state.a / state.b
    return {"result": result}

async def main():
    # Create a workflow graph
    graph = sc.WorkflowGraph(
        state_type=CalculationState,
        enable_scoring=True
    )

    # Add nodes
    graph.add_node("start", lambda s: s, "simple")
    graph.add_node("validate", validate_input, "standard")
    graph.add_node("calculate", perform_calculation, "standard")
    graph.add_node("end", lambda s: s, "simple")

    # Add edges
    graph.add_edge("start", "validate")
    graph.add_edge("validate", "calculate")
    graph.add_edge("calculate", "end")

    # Compile the workflow
    workflow = graph.compile()

    # Create an MCP server with the workflow
    server = create_mcp_server(
        name="WorkflowServer",
        description="MCP server with a calculation workflow",
        workflow=workflow,
        capabilities=["tools", "streaming"],
        enable_scoring=True
    )

    # Run the server
    print("Starting workflow MCP server on http://0.0.0.0:8000")
    await run_mcp_server(server, host="0.0.0.0", port=8000)

if __name__ == "__main__":
    asyncio.run(main())

Next Steps