Skip to content

Pioreactor/mcp-utils

 
 

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

32 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

mcp-utils

A Python utility package for building Model Context Protocol (MCP) servers.

Tests PyPI - Version

Overview

mcp-utils provides utilities and helpers for building MCP-compliant servers in Python, with a focus on synchronous implementations using Flask. This package is designed for developers who want to implement MCP servers in their existing Python applications without the complexity of asynchronous code.

Key Features

  • Basic utilities for MCP server implementation
  • Server-Sent Events (SSE) support
  • Simple decorators for MCP endpoints
  • Synchronous implementation
  • HTTP protocol support
  • SQLite response queue
  • Comprehensive msgspec models for MCP schema
  • Built-in validation and documentation

Installation

Install from this repo

Requirements

  • Python 3.10+
  • msgspec >= 0.18

Optional Dependencies

  • Flask (for web server)

Usage

Basic MCP Server

Here's a simple example of creating an MCP server (using the built-in SQLite queue):

from mcp_utils.core import MCPServer
from mcp_utils.queue import SQLiteResponseQueue
from mcp_utils.schema import GetPromptResult, Message, TextContent, CallToolResult

# Create a basic MCP server with SQLite-backed queue
mcp = MCPServer("example", "1.0", response_queue=SQLiteResponseQueue("responses.db"))

@mcp.prompt()
def get_weather_prompt(city: str) -> GetPromptResult:
    return GetPromptResult(
        description="Weather prompt",
        messages=[
            Message(
                role="user",
                content=TextContent(
                    text=f"What is the weather like in {city}?",
                ),
            )
        ],
    )

@mcp.tool()
def get_weather(city: str) -> str:
    return "sunny"

Flask Example

For production use, you can use a simple Flask app with the mcp server and support Streamable HTTP from version 2025-06-18.

from flask import Flask, jsonify, request
import msgspec
from mcp_utils.core import MCPServer
from mcp_utils.queue import SQLiteResponseQueue

# Create Flask app and MCP server with SQLite-backed queue
app = Flask(__name__)
mcp = MCPServer("example", "1.0", response_queue=SQLiteResponseQueue("responses.db"))

@app.route("/mcp", methods=["POST"])
def mcp_route():
    response = mcp.handle_message(request.get_json())
    # Convert msgspec Struct to builtin types for jsonify
    return jsonify(msgspec.to_builtins(response))


if __name__ == "__main__":
    app.run(debug=True)

For a more comprehensive example including logging setup and session management, check out the example Flask application in the repository.

Running with Gunicorn

Gunicorn is a better approach to running even locally. To run the app with gunicorn

from gunicorn.app.base import BaseApplication

class FlaskApplication(BaseApplication):
    def __init__(self, app, options=None):
        self.options = options or {}
        self.application = app
        super().__init__()

    def load_config(self):
        config = {
            key: value
            for key, value in self.options.items()
            if key in self.cfg.settings
        }
        for key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application


if __name__ == "__main__":
    handler = logging.StreamHandler(sys.stdout)
    formatter = logging.Formatter("[%(asctime)s] [%(levelname)s] %(name)s: %(message)s")
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    options = {
        "bind": "0.0.0.0:9000",
        "workers": 1,
        "worker_class": "gevent",
        "loglevel": "debug",
    }
    FlaskApplication(app, options).run()

Related Projects

  • MCP Python SDK - The official async Python SDK for MCP
  • mcp-proxy - A proxy tool to connect Claude Desktop with MCP servers
  • mcp-utils - Original version with Pydantic support

License

MIT License

Testing with MCP Inspector

The MCP Inspector is a useful tool for testing and debugging MCP servers. It provides a web interface to inspect and test MCP server endpoints.

Installation

Install MCP Inspector using npm:

npm install -g @modelcontextprotocol/inspector

Usage

  1. Start your MCP server (e.g., the Flask example above)
  2. Run MCP Inspector:
git clone [email protected]:modelcontextprotocol/inspector.git
cd inspector
npm run build
npm start
  1. Open your browser and navigate to http://127.0.0.1:6274/
  2. Enter your MCP server URL (e.g., http://localhost:9000/sse)
  3. Use the inspector to:
    • Change transport type to SSE
    • Test server connections
    • Monitor SSE events
    • Send test messages
    • Debug responses

This tool is particularly useful during development to ensure your MCP server implementation is working correctly and complies with the protocol specification.

About

Python utilities to add an MCP server to Flask

Resources

License

Stars

Watchers

Forks

Packages

No packages published

Languages

  • Python 99.9%
  • Shell 0.1%