Rcarmo/umcp: A micro MCP core (asyncio and synchronous)

Rui Carmo released umcp, a zero-dependency Python implementation of the Model Context Protocol (MCP) that supports stdio, Streamable HTTP, legacy SSE, and raw TCP transports. The runtime consists of three files and requires Python 3.10+, with no third-party dependencies. It automatically generates JSON Schema from type hints and infers MCP annotations from naming conventions, enabling dynamic discovery of tools, prompts, and resources.

A lightweight, zero-dependency implementation of the Model Context Protocol MCP https://modelcontextprotocol.io in pure Python -- inspired by the original bash implementation by Muthukumaran Navaneethakrishnan. Why? I found the idea of an MCP server written as a shell script fascinating, and wanted to see what the same idea looked like in Python with proper introspection -- type hints to JSON Schema, naming conventions to MCP annotations, no decorator boilerplate, and the whole thing readable in an afternoon. The runtime is three files: umcp.py , aioumcp.py , and the shared umcp shared.py . It has no third-party runtime dependencies, supports stdio, stateless Streamable HTTP, legacy HTTP+SSE, and raw TCP, and ships with a handful of runnable examples in examples/ /rcarmo/umcp/blob/main/examples . Features -features Requirements -requirements Installation -installation Quick start -quick-start Transports -transports Architecture %EF%B8%8F-architecture Getting started tutorial -getting-started-tutorial Examples -examples Prompt templates -prompt-templates Resources -resources API reference -api-reference Testing -testing Development %EF%B8%8F-development Integration VS Code, Claude Desktop -integration Limitations -limitations Deployment notes -deployment-notes Troubleshooting -troubleshooting Further reading -further-reading License -license - ✅ Full JSON-RPC 2.0 protocol over stdio, SSE, streamable HTTP, or TCP - ✅ Complete MCP protocol implementation tools, prompts, resources , annotations - ✅ Dynamic discovery via function naming convention tool , prompt , resource , resource template - ✅ Runtime tool/prompt/resource registration with list-changed notifications - ✅ Stable sorted discovery with optional cursor pagination for list endpoints - ✅ Complete introspection of function signatures, including Literal , Union , and TypedDict - ✅ MCP inputSchema generated automatically from type hints - ✅ Automatic readOnlyHint / destructiveHint / openWorldHint annotations from naming conventions - ✅ Strict argument validation additionalProperties: false , unknown-arg rejection, type coercion for stringy clients - ✅ Prompt templates for reusable, structured interactions - ✅ Both synchronous and asynchronous implementations -- pick by I/O shape local disk vs. network - ✅ Zero third-party dependencies - Python 3.10+ both bases use PEP 604 unions and types.UnionType git clone https://github.com/rcarmo/umcp cd umcp python -m py compile umcp.py aioumcp.py umcp shared.py No additional packages required -- umcp uses only the Python standard library. echo '{"jsonrpc": "2.0", "method": "tools/call", "params": {"name": "get movies"}, "id": 1}' \ | python ./examples/movie server.py echo '{"jsonrpc": "2.0", "method": "tools/list", "id": 1}' \ | python ./examples/movie server.py echo '{"jsonrpc": "2.0", "method": "tools/call", "params": {"name": "add", "arguments": {"a": 5, "b": 3}}, "id": 1}' \ | python ./examples/calculator server.py stdio remains the default and is the right choice when the MCP host launches the server locally. Plain --port N deliberately retains the old SSE behaviour for compatibility; new network deployments should select Streamable HTTP explicitly. | Transport | Invocation | Intended use | |---|---|---| | stdio | python server.py | Local child-process integrations | | Streamable HTTP | python server.py --port 9000 --http | New network services; stateless POST /mcp | | legacy SSE | python server.py --port 9000 --sse | Existing HTTP+SSE clients; deprecated, with no removal date | | raw TCP | python server.py --port 9000 --tcp | Legacy compatibility only | The equivalent explicit form is --transport stdio|streamable-http|sse|tcp . Network transports also accept --host , --endpoint , --max-request-bytes , and repeatable --allowed-origin options. Conflicting aliases and network transports without --port are rejected. For compatibility with existing lightweight clients and command-line use, the dispatcher does not keep a connection-level "initialized" flag; it will answer methods such as tools/list before initialize . Standards-compliant MCP clients should still perform the normal initialize handshake. Streamable HTTP accepts one JSON-RPC object per POST . Requests return 200 application/json ; notifications and client responses return 202 . The initial initialize negotiates either 2025-03-26 or 2024-11-05 . Later requests must send a supported value in MCP-Protocol-Version . Request targets are matched on the URL path, so /mcp?trace=1 is accepted and treated the same as /mcp . HTTP/1.1 requires exactly one Host header; HTTP/1.0 may omit it. Duplicate Host , Authorization , Origin , Accept , Content-Type , MCP-Protocol-Version , Content-Length , or Transfer-Encoding headers are rejected with 400 , and any Transfer-Encoding is rejected. Stateless GET and DELETE return 405 . Browser preflight is only enabled for an allowed Origin , and only on the configured endpoint path. ┌─────────────┐ ┌───────────────┐ │ MCP Host │ │ MCP Server │ │ AI System │◄──────► │ myserver.py │ └─────────────┘ stdio └───────────────┘ │ ┌─────────┴──────────┐────────────────────┐ ▼ ▼ ▼ ┌────────────────┐ ┌────────────────┐ ┌────────────────────┐ │ Protocol Layer │ │ Business Logic │ │ Prompt Templates │ │ umcp.py │ │ tool methods │ │ prompt methods │ └────────────────┘ └────────────────┘ └────────────────────┘ │ │ ▼ ▼ ┌───────────────┐ ┌───────────────┐ │ Introspection │ │ External │ └───────────────┘ │ Services/APIs │ └───────────────┘ For the design details -- transports, sync vs. async rationale, schema generation, annotation inference, what's deliberately not included -- see docs/ARCHITECTURE.md /rcarmo/umcp/blob/main/docs/ARCHITECTURE.md . Create a file my server.py : python /usr/bin/env python3 from umcp import MCPServer class MyServer MCPServer : """A simple example MCP server.""" def tool greet self, name: str = "World" - str: """Greet someone by name. Args: name: The name to greet Returns: A friendly greeting message """ return f"Hello, {name} " def tool add numbers self, a: float, b: float - float: """Add two numbers together. Args: a: First number b: Second number Returns: The sum of the two numbers """ return a + b if name == " main ": server = MyServer server.run chmod +x my server.py echo '{"jsonrpc": "2.0", "method": "tools/call", "params": {"name": "greet", "arguments": {"name": "Alice"}}, "id": 1}' | ./my server.py echo '{"jsonrpc": "2.0", "method": "tools/call", "params": {"name": "add numbers", "arguments": {"a": 10, "b": 5}}, "id": 2}' | ./my server.py Use AsyncMCPServer from aioumcp instead, and make tool methods async def . Pick async when your tools talk to the network; pick sync when they talk to the local filesystem or run subprocesses. The architecture doc /rcarmo/umcp/blob/main/docs/ARCHITECTURE.md two-implementations-one-shape explains why both exist. python /usr/bin/env python3 import asyncio from aioumcp import AsyncMCPServer class AsyncMyServer AsyncMCPServer : """An async example MCP server.""" async def tool fetch data self, url: str - dict: """Simulate fetching data from a URL.""" await asyncio.sleep 0.1 return {"url": url, "status": "success", "data": "mock response"} if name == " main ": server = AsyncMyServer server.run The runnable examples live under examples/ /rcarmo/umcp/blob/main/examples : -- CRUD over an in-memory store, parameter validation, prompt templates. examples/movie server.py -- pure compute, error handling, type-safe parameters. examples/calculator server.py -- static and templated MCP resources, plus tools that mutate them and emit examples/resource server.py notifications/resources/updated .-- async version of the movie server. examples/async movie server.py -- async version of the calculator. examples/async calculator server.py -- async version of the resource server. examples/async resource server.py Synchronous python examples/movie server.py python examples/calculator server.py Asynchronous python examples/async movie server.py python examples/async calculator server.py For a real, sizeable MCP server built on umcp , see rcarmo/python-office-mcp-server https://github.com/rcarmo/python-office-mcp-server -- a Word/Excel/PowerPoint server with 100+ tools, structured workflow discovery, mutation diagnostics, and the chaining patterns documented in . It's the canonical worked example for what a production deployment of /rcarmo/umcp/blob/main/docs/CHAINING.md docs/CHAINING.md umcp looks like. umcp supports reusable prompt templates using the same naming convention as tools: methods named prompt <name are discovered and exposed via the MCP prompts/list and prompts/get methods. You can also register prompts at runtime with register prompt / unregister prompt , or use register prompt and notify / unregister prompt and notify as convenience wrappers. The plain register/unregister APIs only mutate local state; if you use them directly, you must call notify prompt list changed yourself. See docs/PROMPTS.md /rcarmo/umcp/blob/main/docs/PROMPTS.md for the full reference. Quick example: python class MyServer MCPServer : def prompt code review self, filename: str, issues: int = 0 - str: """Generate a focused code review instruction. Categories: code, review""" return f"Please review '{filename}'. Assume ~{issues} pre-identified issues." echo '{"jsonrpc": "2.0", "method": "prompts/list", "id": 1}' | python ./examples/movie server.py echo '{"jsonrpc": "2.0", "method": "prompts/get", "params": {"name": "code review", "arguments": {"filename": "main.py"}}, "id": 2}' | python ./examples/movie server.py umcp also exposes MCP completion/complete and logging/setLevel . initialize always advertises logging: {} and only adds completions: {} when the server actually has something completable prompt arguments, resource-template arguments, or registered completion providers . Completion values can come from: Literal ... annotations Enum annotations- prompt inputSchema enums supplied to register prompt - registered completion providers via register completion provider Providers receive prefix , arguments , ref , and argument , and may return either a plain list or { "values": ... , "total": N, "hasMore": bool } . Results are prefix-filtered, deduplicated, and capped at 100 values. For runtime logs, call notify log message / log message with one of the standard MCP levels debug , info , notice , warning , error , critical , alert , emergency . Payloads are redacted recursively by default for common secret-looking keys and bearer/token strings; pass sanitize=False to opt out explicitly. Progress and cancellation are request-local. If a client supplies params. meta.progressToken , server code can read it with get progress token and emit notifications/progress via notify progress ... or the server instance method of the same name. No token means no progress notification is sent. Cancellation arrives as notifications/cancelled ; async handlers are cancelled actively when possible, while sync handlers are cooperative only and should call is request cancelled / raise if cancelled inside long-running work. umcp implements the MCP resources spec https://modelcontextprotocol.io/specification/2025-11-25/server/resources : methods named resource <name are exposed as static resources, and methods named resource template <name become parameterised resource templates whose signature parameters fill in the URI placeholders. The MCP capability set is declared automatically on initialize : tools: {"listChanged": true} prompts: {"get": true, "listChanged": true} resources: {"subscribe": true, "listChanged": true} logging: {} completions: {} when prompt/resource-template completion is available The following methods are wired through: tools/list , prompts/list , resources/list , and resources/templates/list return stable sorted results. When called without pagination params they preserve the previous one-shot behaviour. When called with pageSize or a returned cursor they return a page plus nextCursor . resources/list -- returns every resource method, plus anything registered via register resource . resources/templates/list -- every resource template method, plus register resource template . resources/read -- looks up by URI; static resources match by exact URI, templates by {placeholder} regex. Returns -32002 with the URI in data for unknown resources, per spec. resources/subscribe / resources/unsubscribe -- track URIs of interest. Return types are normalised into MCP contents entries: str → text content text + mimeType , default text/plain . bytes → binary content base64-encoded blob , default application/octet-stream . dict → a single content entry, with uri filled in if missing. list dict → multiple content entries. Attach mcp resource = {...} or mcp resource template = {...} to a resource method to override the auto-generated uri / uri template , set a title , description , mime type , size , or annotations audience , priority , lastModified . Mutating tools should call notify resource updated uri only fires for URIs the client has subscribed to and/or notify resource list changed after they change resource state. In the async base these helpers are coroutines await self.notify resource updated ... . Quick example: php class MyServer MCPServer : def resource motd self - str: """Message of the day.""" return "Hello " resource motd. mcp resource = {"mime type": "text/plain", "title": "MOTD"} def resource template user self, user id: str - dict: """Synthesised user profile.""" return {"mimeType": "application/json", "text": f'{{"id": "{user id}"}}'} echo '{"jsonrpc":"2.0","id":1,"method":"resources/list"}' | python examples/resource server.py echo '{"jsonrpc":"2.0","id":2,"method":"resources/read","params":{"uri":"umcp://ResourceServer/motd"}}' | python examples/resource server.py echo '{"jsonrpc":"2.0","id":3,"method":"resources/read","params":{"uri":"file:///hello.txt"}}' | python examples/resource server.py Base class for synchronous MCP servers. discover tools -- finds all tool methods on the subclass. discover prompts -- finds all prompt methods on the subclass. discover resources / discover resource templates -- finds all resource and resource template methods on the subclass. register tool name, callable, ... / unregister tool name -- runtime registration of tools. These are mutation-only; call notify tool list changed explicitly afterwards, or use register tool and notify ... / unregister tool and notify ... . register tool also accepts output schema=... , and tool methods may expose mcp output schema metadata for tools/list . register prompt name, callable, ... / unregister prompt name -- runtime registration of prompts. These are mutation-only; call notify prompt list changed explicitly afterwards, or use register prompt and notify ... / unregister prompt and notify ... . register resource uri, callable, ... / register resource template uri template, callable, ... -- runtime registration of resources/templates. notify tool list changed / notify prompt list changed / notify resource updated uri / notify resource list changed -- emit MCP notifications when the advertised catalogue changes. In AsyncMCPServer , these notification helpers are coroutines. handle tools call -- dispatches a tool call. Mapping / structured tool results preserve the legacy text content block and also expose structuredContent when possible; advertised outputSchema values are validated on the way out. handle prompt get -- dispatches a prompt fetch. get config -- override to declare server name, version, capabilities. get instructions -- override to give the model session-level guidance. run -- start the server on the configured transport stdio by default; plain --port N retains legacy SSE, --http selects Streamable HTTP, and --tcp selects raw TCP . authenticate request / authorize request -- optional HTTP identity and policy hooks. The default principal is anonymous for compatibility.- Module-level umcp.get request context -- return immutable request-local transport, protocol, principal, peer, and header metadata. The same MCP feature surface, with coroutine entry points named process request async , run socket async , run sse async , and run streamable http async ; tool and prompt methods may also be async def . Use this when your tools are network-bound; use MCPServer when they're local-disk or compute-bound. Remote services can validate HTTP credentials without turning identity into a model-controlled tool argument: python from typing import Mapping from umcp import MCPServer, MCPPrincipal, get request context class PrivateServer MCPServer : def authenticate request self, , method: str, path: str, headers: Mapping str, str , peer: str | None, - MCPPrincipal | None: if headers.get "authorization" = "Bearer expected-token": return None return MCPPrincipal name="alice", roles= "reader", def authorize request self, principal: MCPPrincipal | None, , rpc method: str | None, tool name: str | None, - bool: return principal is not None and tool name = "write" or "writer" in principal.roles def tool whoami self - str: return get request context .principal or "local" Authentication failure is an HTTP 401 ; authorization failure is 403 . The context is reset in finally after every dispatch and is isolated across threads, asyncio tasks, and executor workers. MCPPrincipal.metadata and MCPRequestContext.headers are defensive immutable copies. php def tool <name self, param1: type1, param2: type2 = default - return type: """Tool description first line becomes the summary . Args: param1: Description of parameter 1 param2: Description of parameter 2 Returns: Description of return value """ php def prompt <name self, param1: type1, param2: type2 = default - return type: """Prompt description. Categories: category1, category2""" For schema generation rules Literal , Union , Optional , TypedDict and annotation inference, see docs/ARCHITECTURE.md /rcarmo/umcp/blob/main/docs/ARCHITECTURE.md . The project uses pytest and ships a Makefile plus a GitHub Actions workflow that runs the suite on Python 3.10, 3.11, and 3.12. Make targets preferred make test full suite make test-fast -x -q make coverage coverage report on stdout make coverage-html writes htmlcov/index.html make clean remove caches and log files Or pytest directly python -m pytest tests/ Or with uv hermetic, no global pytest required uv run --with pytest --with pytest-asyncio --python 3.12 \ python -m pytest tests/ Specific files / verbose python -m pytest tests/test resources.py -v | File | Area | |---|---| test introspection.py | end-to-end tool discovery via subprocess | test movieserver.py | worked-example end-to-end test | test protocol errors.py | JSON-RPC error paths -- parse/version/method/args/raise | test schema generation.py | type hint - JSON Schema mapping primitives, Optional , Union , Literal , TypedDict , list / dict | test schema fallbacks.py | exotic-type schema fallback behaviour | test annotations.py | readOnlyHint / destructiveHint / openWorldHint inference + overrides | test coercion.py | stringy-client argument coercion str - int / float / bool | test prompts.py / test async prompts.py | prompt discovery and dispatch | test resources.py | resources/list, /templates/list, /read, subscribe/unsubscribe, dynamic registration, capability declaration | test notifications.py | notifications/resources/list changed and notifications/resources/updated emission and subscription gating | test async servers.py | async stdio subprocess round-trips | test transports.py | sync stdio, TCP, legacy SSE, and Streamable HTTP subprocess tests | test shared negotiation context.py | protocol negotiation and immutable request/principal context | test streamable http sync async.py | matching sync/async Streamable HTTP dispatch | test streamable http regressions.py | auth, Origin, CORS, limits, context isolation, CLI, and remote-safe errors | simple async test.py | smoke tests for the async base | The suite covers both bases and runs on Python 3.10, 3.11, and 3.12. git clone https://github.com/rcarmo/umcp cd umcp python -m pytest tests/ - Explicit imports only. - Functional style; short, single-responsibility functions. - Type hints on all parameters and returns. - Double quotes for strings; triple-double-quote docstrings. snake case method naming.- f-strings only when needed. - Logging over print statements. - Fork the repository. - Create a feature branch: git checkout -b feature-name . - Make your changes following the code style. - Add tests for new functionality. - Ensure all tests pass: python -m pytest tests/ . - Submit a pull request. umcp/ ├── umcp.py -- sync MCPServer base class ├── aioumcp.py -- async AsyncMCPServer base class ├── umcp shared.py -- versions, principals, and request context ├── scripts/ -- optional compatibility smoke tests ├── examples/ -- runnable example servers │ ├── movie server.py │ ├── async movie server.py │ ├── calculator server.py │ ├── async calculator server.py │ ├── resource server.py │ └── async resource server.py ├── tests/ -- pytest suite ├── docs/ │ ├── ARCHITECTURE.md -- design, transports, schema generation │ ├── CHAINING.md -- chaining patterns for MCP server authors │ └── PROMPTS.md -- prompt template reference ├── readme.md -- this file └── LICENSE "mcp": { "servers": { "my-weather-server": { "type": "stdio", "command": "/path/to/your/server.py", "args": , "env": { "MCP API KEY": "anything you need" } } } } Then /mcp my-weather-server get weather for New York from Copilot Chat. { "mcpServers": { "my-server": { "command": "python", "args": "/path/to/your/server.py" , "env": {} } } } - No streaming responses -- partial results aren't supported. - No built-in authentication backend or policy engine. Streamable HTTP exposes authenticate request and authorize request hooks, and reverse proxies can supply trusted identity, but the application owns credentials, roles, and policy. - No concurrency in the synchronous version -- one request at a time. Use AsyncMCPServer if you need overlapping I/O. For most AI-assistant / local-tool use cases, none of these are blocking. See docs/ARCHITECTURE.md /rcarmo/umcp/blob/main/docs/ARCHITECTURE.md what-umcp-deliberately-doesnt-do for the rationale. For local hosts Claude Desktop, VS Code, Copilot, Piclaw, etc. , prefer stdio . umcp now tags stdio/file requests as local request contexts, keeps request metadata immutable, and avoids exposing internal exception strings over remote transports. If you need network access, prefer streamable HTTP over legacy SSE/TCP: - bind to loopback unless you have a real reason not to; - implement authenticate request and authorize request for remote use; hook exceptions are logged server-side and returned as generic 500 s; - require a supported MCP-Protocol-Version on non- initialize requests; - set --allowed-origin explicitly for browser clients, or rely on loopback-only Origin handling for local web UIs; - expect OPTIONS preflight only when the browser sends a valid Origin . A typical reverse-proxy setup should preserve Host , Authorization , Origin , Accept , Content-Type , and MCP-Protocol-Version headers unchanged, and should not inject duplicates or Transfer-Encoding . If the proxy terminates TLS or auth, keep the upstream umcp listener on localhost. Minimal systemd service sketch: Unit Description=umcp HTTP server After=network.target Service ExecStart=/usr/bin/python /srv/umcp/server.py --port 9000 --http --host 127.0.0.1 --allowed-origin https://ui.example WorkingDirectory=/srv/umcp Restart=on-failure User=umcp Group=umcp Install WantedBy=multi-user.target Container deployments should follow the same pattern: keep the container port private, expose it through a reverse proxy, and do not publish raw SSE/TCP unless you have a separate auth story. Session IDs, if stateful HTTP support is added later, are routing identifiers rather than credentials. Piclaw's pi-mcp-adapter uses the official MCP SDK's StreamableHTTPClientTransport before considering its legacy SSE fallback. The repository includes a repeatable smoke test for that exact transport: Terminal 1 python examples/calculator server.py --port 9000 --http Terminal 2 -- point this at Piclaw's bundled SDK directory MCP URL=http://127.0.0.1:9000/mcp \ MCP SDK ROOT=/path/to/piclaw/node modules/@modelcontextprotocol/sdk \ bun scripts/piclaw streamable http smoke.ts A successful run connects without SSE, discovers the calculator tools, and calls add 2, 3 . This is an optional compatibility check; Bun and the MCP SDK are not runtime dependencies of umcp . Server doesn't respond to JSON-RPC requests. Check the JSON is valid and the server is running. Try a tools/list request first -- it has no arguments and exercises the protocol path. Tools not showing up in tools/list. Ensure the methods are named tool and have proper type hints. Check mcpserver.log next to the script for introspection errors. Async server seems slow. The async examples use asyncio.sleep to simulate I/O. Remove these in real applications. Permission denied on the script. chmod +x your server.py . Debug logging. if name == " main ": server = MyServer server.logger.setLevel "DEBUG" server.run -- design notes: transports, sync vs. async rationale, schema generation, annotation inference, what's deliberately not included. docs/ARCHITECTURE.md -- how language models actually chain MCP tool calls in practice, with docs/CHAINING.md as the worked example. Read this before building anything non-trivial; it's where the real reliability work lives. python-office-mcp-server -- prompt template reference. docs/PROMPTS.md -- production-grade MCP server built on python-office-mcp-server umcp , with 100+ tools for Word/Excel/PowerPoint editing. MIT -- see LICENSE /rcarmo/umcp/blob/main/LICENSE . - Inspired by the original bash MCP implementation by Muthukumaran Navaneethakrishnan. - Built against the Model Context Protocol https://modelcontextprotocol.io specification.