Model Context Protocol SDK (Python Implementation)

This repository contains the Python SDK for implementing the Model Context Protocol (MCP), designed to decouple LLM interaction logic from context provisioning.

Setup & Execution

Ensure you have Python configured correctly:

which python
python3 -m venv service_env
source service_env/Scripts/activate
pip install -r requirements.txt
python fast_mcp_server.py

Operational Modes

Client Initialization

To prepare the environment for client testing:

# Configure API credentials
.env
OPENAI_API_KEY="your_secret_key_here"

Server Hosting

Launch the primary application endpoint:

uvicorn ne04j_mcp_server:app --host 0.0.0.0 --port 8000

Data Modeling (Neo4j Cypher Examples)

Introducing Data Entities (Nodes)

CREATE (p1:Person {name: "Tom Hanks", birthYear: 1956})
CREATE (p2:Person {name: "Kevin Bacon", birthYear: 1958})
CREATE (m1:Movie {title: "Forrest Gump", releaseYear: 1994})
CREATE (m2:Movie {title: "Apollo 13", releaseYear: 1995})

Defining Relationships (Edges)

MATCH (p:Person {name: "Tom Hanks"}), (m:Movie {title: "Forrest Gump"})
CREATE (p)-[:ACTED_IN]->(m)

MATCH (p:Person {name: "Tom Hanks"}), (m:Movie {title: "Apollo 13"})
CREATE (p)-[:ACTED_IN]->(m)

MATCH (p1:Person {name: "Tom Hanks"}), (p2:Person {name: "Kevin Bacon"})
CREATE (p1)-[:FRIENDS_WITH]->(p2)

Attribute Augmentation

MATCH (p:Person {name: "Tom Hanks"})
SET p.oscarsWon = 2

MATCH (m:Movie {title: "Forrest Gump"})
SET m.genre = "Drama"

MATCH (p:Person {name: "Tom Hanks"})-[r:ACTED_IN]->(m:Movie {title: "Forrest Gump"})
SET r.role = "Forrest Gump"

Constructing Hierarchical Structures

// Establish a group and link individuals to it
CREATE (c1:Community {name: "Hollywood Stars"})
WITH c1
MATCH (p:Person)
WHERE p.name IN ["Tom Hanks", "Kevin Bacon"]
CREATE (p)-[:MEMBER_OF]->(c1)

Boilerplate & Project Metadata

Contents

• Introduction
• Acquisition
• First Steps
• What is MCP?
• Fundamental Components
• Server Host
• Data Accessors (Resources)
• Action Handlers (Tools)
• Instruction Templates (Prompts)
• Media Objects (Images)
• Execution Context
• Deployment Strategies
• Rapid Prototyping
• Desktop Integration
• Direct Invocation
• Illustrative Examples
• Pass-Through Utility
• Database Interrogator
• Advanced Topics
• Unmediated Server Control
• Client SDK Usage
• Protocol Building Blocks
• Initialization Declarations
• Reference Materials
• Contribution Guidelines
• Licensing

Overview

The Model Context Protocol standardizes how contextual artifacts (data, functions, templates) are supplied to Language Models, decoupling the provider from the ultimate consumer. This Python Software Development Kit implements the entire MCP specification, enabling:

• Development of universal MCP consumers capable of interfacing with any compliant server.
• Construction of MCP endpoints that advertise resources, functions, and contextual prompts.
• Utilization of conventional communication channels such as standard I/O and Server-Sent Events (SSE).
• Automated management of all protocol messaging and state transitions.

Acquisition

We endorse the use of uv for managing Python environments:

uv add "mcp[cli]"

Alternatively, standard installation is supported:

pip install mcp

First Steps

Let's deploy a minimal MCP host exposing arithmetic logic and configuration data:

# host_app.py
from mcp.server.fastmcp import FastMCP

# Instantiate the server instance
mcp = FastMCP("DemonstrationHost")

# Define an addition utility function
@mcp.tool()
def add(a: int, b: int) -> int:
    """Calculates the sum of two integers."""
    return a + b

# Define a parameterized data retrieval service
@mcp.resource("config://user/{user_id}")
def get_user_config(user_id: str) -> str:
    """Fetches specific user configuration settings."""
    return f"User {user_id} settings loaded."

This host can be integrated directly into platforms like Claude Desktop via:

mcp install host_app.py

For interactive debugging, use the Inspector utility:

mcp dev host_app.py

What is MCP?

The Model Context Protocol (MCP) establishes a secure, standardized mechanism for servers to offer encapsulated data and executable functions to LLM consumption pipelines. It functions akin to a specialized Web API for LLM contexts. MCP hosts can:

• Offer data via Resources (analogous to idempotent GET operations; used for context infusion).
• Expose executable logic via Tools (analogous to operations with side-effects; used for computation or external invocation).
• Define structured interaction flows via Prompts (pre-validated conversational blueprints).
• And other standardized capabilities.

Fundamental Components

Server Host

The FastMCP host class manages protocol adherence, connection lifecycles, and message routing:

from dataclasses import dataclass
from typing import AsyncIterator
from mcp.server.fastmcp import FastMCP

# Initialize server with a distinct identifier
mcp = FastMCP("Orchestrator")

# Declare external dependencies required for runtime
mcp = FastMCP("Orchestrator", dependencies=["sqlalchemy", "redis"])

@dataclass
class OperationalState:
    db_pool: DatabaseConnectionPool

@asynccontextmanager
async def server_startup_handler(server: FastMCP) -> AsyncIterator[OperationalState]:
    """Manages state initialization and teardown."""
    try:
        # Establish external connections
        await db_pool.initialize()
        yield OperationalState(db_pool=db_pool)
    finally:
        # Release resources
        await db_pool.terminate()

# Inject lifespan management
mcp = FastMCP("Orchestrator", lifespan=server_startup_handler)

# Access state context within handlers
@mcp.tool()
def perform_transaction(ctx: Context) -> str:
    """Tool utilizing pre-initialized database access."""
    pool = ctx.request_context.lifespan_context["db_pool"]
    return pool.execute_transaction()

Data Accessors (Resources)

Resources provide read-only data streams to the LLM context, analogous to idempotent API retrieval methods:

@mcp.resource("metadata://system_params")
def get_system_metadata() -> str:
    """Static global parameters."""
    return "Version: 1.0; Mode: Production"

@mcp.resource("user_data/details/{user_id}")
def retrieve_user_details(user_id: str) -> str:
    """Retrieves granular user information based on identifier."""
    return f"Data for user ID: {user_id}"

Action Handlers (Tools)

Tools expose encapsulated functionality to the LLM, suitable for performing computations or creating observable side-effects:

@mcp.tool()
def process_payment(amount_usd: float, token: str) -> bool:
    """Executes a financial transaction."""
    # ... external payment API call ...
    return True

@mcp.tool()
async def retrieve_external_feed(endpoint: str) -> str:
    """Fetches dynamic content from a specified URL."""
    async with httpx.AsyncClient() as client:
        response = await client.get(endpoint)
        return response.text

Instruction Templates (Prompts)

Prompts serve as predefined, reusable conversational structures that guide LLM input formatting:

@mcp.prompt()
def summarize_document(text_blob: str) -> str:
    return f"Condense the following text into three bullet points:\n{text_blob}"

@mcp.prompt()
def escalate_ticket(issue_description: str) -> list[Message]:
    return [
        UserMessage("Urgent escalation request."),
        AssistantMessage("Details provided:"),
        UserMessage(issue_description),
        AssistantMessage("Routing to Tier 3 support now.")
    ]

Media Objects (Images)

FastMCP includes utility for native handling of image data via the Image wrapper:

from mcp.server.fastmcp import Image
from PIL import Image as PILImage

@mcp.tool()
def generate_icon(source_file: str) -> Image:
    """Reduces an input image to a standard icon size."""
    img = PILImage.open(source_file)
    img.thumbnail((64, 64))
    return Image(data=img.tobytes(), format="jpeg")

Execution Context

The Context object provides introspection and interaction mechanisms to handlers:

from mcp.server.fastmcp import Context

@mcp.tool()
async def batch_process(item_list: list[str], ctx: Context) -> str:
    """Handles large operations with reporting."""
    total = len(item_list)
    for index, item in enumerate(item_list):
        ctx.log_debug(f"Processing item {item}")
        await ctx.report_progress(index + 1, total)
        # Access external data if needed
        data, _ = await ctx.read_resource(f"external/{item}")
    return "Batch concluded successfully."

Deployment Strategies

Rapid Prototyping

The fastest method for immediate testing utilizes the MCP Inspector utility:

mcp dev host_app.py

# Including required external libraries
mcp dev host_app.py --with pandas --with requests

# Enabling hot-reloading for local source code
mcp dev host_app.py --with-editable .

Desktop Integration

For persistent use within desktop environments:

mcp install host_app.py

# Assigning a custom moniker
mcp install host_app.py --name "My Internal Utility Service"

# Passing runtime configuration variables
mcp install host_app.py -v SECRET=xxxx -v ENDPOINT=prod.svc
mcp install host_app.py -f .env.production

Direct Invocation

For self-contained execution environments:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Standalone Host")

if __name__ == "__main__":
    mcp.run()

Execute via:

python host_app.py
# OR
mcp run host_app.py

Illustrative Examples

Pass-Through Utility

A minimal implementation demonstrating all core interfaces:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("EchoService")

@mcp.resource("echo/data/{input_val}")
def resource_echo(input_val: str) -> str:
    """Reflects input via a Resource access."""
    return f"Resource reflection: {input_val}"

@mcp.tool()
def tool_echo(input_val: str) -> str:
    """Reflects input via a Tool execution."""
    return f"Tool execution result: {input_val}"

@mcp.prompt()
def prompt_echo(input_val: str) -> str:
    """Defines a standardized input wrapper."""
    return f"LLM instruction based on input: {input_val}"

Database Interrogator

An example demonstrating safe interaction with an SQLite backend:

from mcp.server.fastmcp import FastMCP
import sqlite3

mcp = FastMCP("SQLBridge")

@mcp.resource("db_meta/structure")
def get_schema() -> str:
    """Exposes the database table definitions."""
    conn = sqlite3.connect("data.db")
    schema_dumps = conn.execute(
        "SELECT sql FROM sqlite_master WHERE type='table'"
    ).fetchall()
    return "\n".join(sql[0] for sql in schema_dumps if sql[0])

@mcp.tool()
def execute_safe_query(sql_command: str) -> str:
    """Permits read-only SQL execution."""
    conn = sqlite3.connect("data.db")
    try:
        result_set = conn.execute(sql_command).fetchall()
        return "\n".join(str(row) for row in result_set)
    except Exception as err:
        return f"Execution Failure: {str(err)}"

Advanced Topics

Unmediated Server Control

For scenarios demanding granular control over protocol framing and lifecycle, utilize the underlying Server class:

from contextlib import asynccontextmanager
from typing import AsyncIterator

@asynccontextmanager
async def custom_lifespan(server: Server) -> AsyncIterator[dict]:
    """Manual lifecycle hook management."""
    try:
        # Initialization logic
        await db_client.start()
        yield {"db": db_client}
    finally:
        # Cleanup logic
        await db_client.stop()

server = Server("fine-grained-host", lifespan=custom_lifespan)

# Accessing context during a function call
@server.call_tool()
async def fetch_record(name: str, args: dict) -> list:
    context_data = server.request_context
    db_handle = context_data.lifespan_context["db"]
    return await db_handle.fetch(args["id"])

The lifespan API offers: - A mechanism for initializing persistent assets upon host commencement and ensuring orderly release upon termination. - Direct forwarding of initialized state objects into the request handling scope. - Strong typing support for context propagation across lifespan boundaries.

from mcp.server.lowlevel import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types

# Instantiate the barebones server object
server = Server("stdio-example")

@server.list_prompts()
async def provide_prompt_manifest() -> list[types.Prompt]:
    return [
        types.Prompt(
            name="config_query",
            description="A template to query system settings",
            arguments=[
                types.PromptArgument(
                    name="setting_key",
                    description="The parameter name to fetch",
                    required=True
                )
            ]
        )
    ]

@server.get_prompt()
async def resolve_prompt_template(
    name: str,
    arguments: dict[str, str] | None
) -> types.GetPromptResult:
    if name != "config_query":
        raise ValueError(f"Template unknown: {name}")

    return types.GetPromptResult(
        description="Retrieved configuration query",
        messages=[
            types.PromptMessage(
                role="system",
                content=types.TextContent(
                    type="text",
                    text="Acknowledge request and retrieve setting."
                )
            )
        ]
    )

async def bootstrap():
    async with mcp.server.stdio.stdio_server() as (reader, writer):
        await server.run(
            reader,
            writer,
            InitializationOptions(
                server_name="low-level-io",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                )
            )
        )

if __name__ == "__main__":
    import asyncio
    asyncio.run(bootstrap())

Client SDK Usage

The SDK furnishes a high-level abstraction layer for interacting with remote MCP hosts:

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import CreateMessageRequestParams, CreateMessageResult

# Define connection parameters for an execution process
server_params = StdioServerParameters(
    command="python", 
    args=["remote_host.py"], 
    env=None 
)

# Optional: Define a fallback model response handler
async def fallback_model_response(message: CreateMessageRequestParams) -> CreateMessageResult:
    return CreateMessageResult(
        role="assistant",
        content=types.TextContent(
            type="text",
            text="[Fallback Response: System Message Received]",
        ),
        model="local-fallback",
        stopReason="fallback_handler",
    )

async def establish_session():
    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream, sampling_callback=fallback_model_response) as session:
            # Initiate protocol handshake
            await session.initialize()

            # Query advertised functions
            available_tools = await session.list_tools()

            # Retrieve specific template
            template = await session.get_prompt("config_query", arguments={"setting_key": "timeout"})

            # Access remote data endpoint
            data_payload, mime = await session.read_resource("file:///etc/config")

            # Trigger remote execution
            tool_result = await session.call_tool("data_processor", arguments={"id": 101})

if __name__ == "__main__":
    import asyncio
    asyncio.run(establish_session())

Protocol Building Blocks

The MCP specification relies on three core interface categories for host functionality:

Initialization Declarations

MCP hosts communicate their functional scope during the initial connection phase:

Reference Materials

• Official Model Context Protocol Documentation
• Canonical MCP Specification Document
• Registry of Community-Built Hosts

Contribution Guidelines

We welcome contributions from all skill levels. Consult the CONTRIBUTING.md file for guidelines on submitting enhancements or fixes.

Licensing

This software is distributed under the permissive MIT License; see the LICENSE file for complete terms.