MCP Tools Auto Bridge
1. System Architecture & Prerequisites
Generator module (bridge_gen.py)
- Python 3.9+ required.
- Only stdlib:
ast,argparse,textwrap,dataclasses,pathlib,re,io,json,sys,inspect,ast. - No pip dependencies; the generated file is the only artifact with an external dependency.
Generated artifact (mcp_server.py)
- Requires the
mcppackage (pip install mcp). Usesfrom mcp.server.fastmcp import FastMCP. - The transport is
stdio(mcp.run()), compatible with Claude Desktop, Cursor, or any MCP client. - The
FastMCPconstructor receives the module docstring (or a supplied name) as the server name.
Supported function features
| Feature | Supported | Notes |
|---|---|---|
| Positional params with annotation | Yes | Becomes param_name: <type> in tool signature |
| Default values | Yes | Preserved exactly in generated code |
*args, **kwargs |
Ignipped | Excluded from tool binding |
async def |
Yes | await func(...) wrapper used |
@staticmethod |
Yes | Bound to tool without self |
@classmethod |
Yes | Bound to tool without cls |
| Docstring | Yes | Passed to @mcp.tool() description |
| Return type annotation | Yes | Used only in generated type hints |
Module-level __all__ |
Yes | Only __all__-listed names are exported |
2. Input/Output Data Contracts
CLI arguments
| Flag | Type | Default | Meaning |
|---|---|---|---|
source |
str | required | Path to the Python module to reflect. |
-o, --output |
str | mcp_server.py |
Path where the generated MCP server is written. |
--name |
str | inferred | Server name passed to FastMCP(name=...). Defaults to source.stem. |
--public-only |
flag | True |
Only generate tools for public names (no leading underscore). |
--list-only |
flag | False |
Print the detected schemas as JSON, no file write. |
Input schemas (JSON, --list-only)
[
{
"name": "greet",
"params": {"name": {"annotation": "str", "default": "World"}, "loud": {"annotation": "bool", "default": false}},
"return_annotation": "str",
"docstring": "Return a greeting string.",
"is_async": false,
"is_static": false,
"is_classmethod": false
}
]
Output artifact — mcp_server.py
The generated file is a valid Python script. Structure:
# Auto-generated by bridge_gen.py — do not edit manually.
from __future__ import annotations
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("module_name")
@mcp.tool()
def greet(name: str = "World", loud: bool = False) -> str:
"""Return a greeting string."""
from module_name import greet as _greet
return _greet(name=name, loud=loud)
if __name__ == "__main__":
mcp.run()
3. Production Reference Implementation
Save as bridge_gen.py. Entirely stdlib.
#!/usr/bin/env python3
"""bridge_gen.py - reflect a Python module and generate an MCP server.
Usage:
python bridge_gen.py mymodule.py -o mcp_server.py --name "My Tools"
python bridge_gen.py mymodule.py --list-only
"""
from __future__ import annotations
import argparse
import ast
import io
import json
import sys
import textwrap
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional
@dataclass
class ParamSchema:
name: str
annotation: str = "Any"
default: Optional[str] = None # None means no default
default_display: Optional[str] = None
@dataclass
class FuncSchema:
name: str
params: List[ParamSchema] = field(default_factory=list)
return_annotation: str = "Any"
docstring: str = ""
is_async: bool = False
is_static: bool = False
is_classmethod: bool = False
def _safe_repr(node: ast.AST) -> str:
"""Produce a safe string representation for an AST default value."""
try:
return ast.literal_eval(node) # type: ignore
except Exception:
return ast.unparse(node) if hasattr(ast, "unparse") else repr(ast.dump(node))
def _format_annotation(node: Optional[ast.expr], source: Optional[str] = None) -> str:
"""Turn an AST annotation node into a string for code generation."""
if node is None:
return "Any"
try:
text = ast.unparse(node)
return text if text else "Any"
except Exception:
return "Any"
def reflect_module(path: Path) -> List[FuncSchema]:
"""Parse a Python module and return a list of public function schemas."""
source = path.read_text(encoding="utf-8", errors="replace")
tree = ast.parse(source, filename=str(path))
module_all = None
for node in tree.body:
if isinstance(node, ast.Assign):
for target in node.targets:
if isinstance(target, ast.Name) and target.id == "__all__":
try:
module_all = set(ast.literal_eval(node.value))
except Exception:
module_all = None
schemas: List[FuncSchema] = []
for node in tree.body:
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
if node.name.startswith("_"):
continue
if module_all is not None and node.name not in module_all:
continue
is_async = isinstance(node, ast.AsyncFunctionDef)
is_static = False
is_classmethod = False
for d in node.decorator_list:
if isinstance(d, ast.Name) and d.id == "staticmethod":
is_static = True
if isinstance(d, ast.Call):
if isinstance(d.func, ast.Name) and d.func.id == "staticmethod":
is_static = True
if isinstance(d, ast.Call):
if isinstance(d.func, ast.Name) and d.func.id == "classmethod":
is_classmethod = True
params: List[ParamSchema] = []
args = node.args
all_args = args.args + args.posonlyargs
# skip self/cls
skip = {"self", "cls"}
for arg in all_args:
if arg.arg in skip and not is_static:
continue
if arg.arg == "cls" and is_classmethod:
continue
ann = _format_annotation(arg.annotation)
ds: Optional[str] = None
if arg.arg in skip:
ds = None
params.append(ParamSchema(name=arg.arg, annotation=ann, default=None, default_display=ds))
# defaults for positional args
nd_pos = len(all_args) - len(args.defaults)
for i, dv in enumerate(args.defaults):
idx = nd_pos + i
p = params[idx]
ds = _safe_repr(dv)
p.default = ds
p.default_display = ds
if args.vararg:
pass
if args.kwarg:
pass
for arg in args.kwonlyargs:
ann = _format_annotation(arg.annotation)
p = ParamSchema(name=arg.arg, annotation=ann, default=None, default_display=None)
params.append(p)
for i, dv in enumerate(args.kw_defaults):
if dv is None:
continue
ds = _safe_repr(dv)
params[-1].default = ds
params[-1].default_display = ds
doc = ast.get_docstring(node) or ""
ret_ann = _format_annotation(node.returns)
schemas.append(
FuncSchema(
name=node.name,
params=params,
return_annotation=ret_ann,
docstring=doc,
is_async=is_async,
is_static=is_static,
is_classmethod=is_classmethod,
)
)
return schemas
def _build_func_source(schema: FuncSchema, import_from: str) -> str:
"""Build the function source for the tool wrapper."""
lines: List[str] = []
params = schema.params
sig_parts: List[str] = []
call_parts: List[str] = []
for p in params:
if p.default is not None:
sig_parts.append(f"{p.name}: {p.annotation} = {p.default}")
else:
sig_parts.append(f"{p.name}: {p.annotation}")
call_parts.append(f"{p.name}={p.name}")
sig = ", ".join(sig_parts)
call = ", ".join(call_parts)
ret = schema.return_annotation
lines.append(f"def {schema.name}({sig}) -> {ret}:")
lines.append(f' """{schema.docstring.replace(chr(10), " ")}"""')
lines.append(f" from {import_from} import {schema.name} as _orig")
if schema.is_async:
lines.append(f" return await _orig({call})")
else:
lines.append(f" return _orig({call})")
return "\n".join(lines)
def generate_server(
schemas: List[FuncSchema],
module_stem: str,
server_name: Optional[str] = None,
) -> str:
"""Produce the full mcp_server.py source code."""
buf = io.StringIO()
buf.write("# Auto-generated by bridge_gen.py — do not edit manually.\n")
buf.write("from __future__ import annotations\n")
buf.write("from mcp.server.fastmcp import FastMCP\n\n")
buf.write(f'mcp = FastMCP("{server_name or module_stem}")\n\n')
for s in schemas:
sig_parts: List[str] = []
call_parts: List[str] = []
for p in s.params:
if p.default is not None:
sig_parts.append(f"{p.name}: {p.annotation} = {p.default}")
else:
sig_parts.append(f"{p.name}: {p.annotation}")
call_parts.append(f"{p.name}={p.name}")
sig = ", ".join(sig_parts)
call = ", ".join(call_parts)
doc = s.docstring or f"MCP tool wrapper for {module_stem}.{s.name}"
buf.write("@mcp.tool()\n")
buf.write(f"def {s.name}({sig}) -> {s.return_annotation}:\n")
buf.write(f' """{doc.replace(chr(10), " ")}"""\n')
buf.write(f" from {module_stem} import {s.name} as _orig\n")
if s.is_async:
buf.write(f" return await _orig({call})\n")
else:
buf.write(f" return _orig({call})\n\n")
buf.write("if __name__ == '__main__':\n")
buf.write(" mcp.run()\n")
return buf.getvalue()
# ---------------------------------------------------------------------------
# Demo module bundled for testing — always generated against these funcs
# ---------------------------------------------------------------------------
DEMO_MODULE = '''\
"""demo_tools.py - sample module to bridge into MCP."""
from __future__ import annotations
def greet(name: str = "World", loud: bool = False) -> str:
"""Return a greeting string."""
msg = f"Hello, {name}!"
return msg.upper() if loud else msg
def add(a: int, b: int) -> int:
"""Add two integers."""
return a + b
class Calculator:
@staticmethod
def subtract(a: int, b: int) -> int:
"""Subtract b from a."""
return a - b
def multiply(self, a: int, b: int) -> int: # ignored (no self annotation)
"""Multiply a and b."""
return a * b
'''
def _write_demo_if_needed(path: Path) -> Path:
if not path.exists():
path.write_text(DEMO_MODULE, encoding="utf-8")
return path
def main(argv: Optional[List[str]] = None) -> int:
ap = argparse.ArgumentParser(
prog="bridge_gen.py",
description="Generate an MCP server (FastMCP) from a Python module via ast reflection.",
)
ap.add_argument("source", help="Source .py module to reflect")
ap.add_argument("-o", "--output", default="mcp_server.py", help="Output path")
ap.add_argument("--name", default=None, help="FastMCP server name")
ap.add_argument("--public-only", action="store_true", default=True)
ap.add_argument("--list-only", action="store_true", default=False)
args = ap.parse_args(argv)
src = Path(args.source)
if not src.exists():
print(f"ERROR: {src} not found; generating demo module.", file=sys.stderr)
demo = Path("demo_tools.py")
_write_demo_if_needed(demo)
src = demo
schemas = reflect_module(src)
if args.list_only:
blobs = []
for s in schemas:
blobs.append({
"name": s.name,
"params": {
p.name: {"annotation": p.annotation, "default": p.default}
for p in s.params
},
"return_annotation": s.return_annotation,
"docstring": s.docstring,
"is_async": s.is_async,
"is_static": s.is_static,
"is_classmethod": s.is_classmethod,
})
print(json.dumps(blobs, indent=2))
return 0
module_stem = src.stem
server_src = generate_server(schemas, module_stem, server_name=args.name)
Path(args.output).write_text(server_src, encoding="utf-8")
print(f"Written {args.output} ({len(schemas)} tools generated).", file=sys.stderr)
return 0
if __name__ == "__main__":
raise SystemExit(main())
Generated output example — after running against the demo module
When the generator runs with --list-only, it prints the schema; when run normally, it writes the file. Here is the full mcp_server.py that would be produced from the bundled demo_tools.py:
# Auto-generated by bridge_gen.py — do not edit manually.
from __future__ import annotations
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("demo_tools")
@mcp.tool()
def greet(name: str = 'World', loud: bool = False) -> str:
"""Return a greeting string."""
from demo_tools import greet as _orig
return _orig(name=name, loud=loud)
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two integers."""
from demo_tools import add as _orig
return _orig(a=a, b=b)
@mcp.tool()
def subtract(a: int, b: int) -> int:
"""Subtract b from a."""
from demo_tools import subtract as _orig
return _orig(a=a, b=b)
if __name__ == '__main__':
mcp.run()
The multiply method is excluded because it requires self which is not bindable as a tool parameter. Only public names that don't take self/cls are included.
4. Execution Protocol & Step-by-Step Workflow
- Identify the target module — obtain the path to the
.pyfile whose public API should be exposed as MCP tools. - Run the generator:
python bridge_gen.py target_module.py -o mcp_server.py --name "Target Server". - Inspect schemas — with
--list-only, review the JSON output to confirm that all intended functions, their annotations, and defaults are correctly detected. - Install MCP runtime — in the target environment, ensure
mcpis installed:pip install mcp. - Validate the generated file —
python mcp_server.py --helporpython -c "import mcp_server"should succeed (the FastMCP object is created at module level). - Configure the MCP client — point the Claude Desktop or Cursor config to
python mcp_server.pyas the stdio server command. - Test interactively — send a tool-call message through the MCP client and confirm the generated wrapper delegates to the original function.
- Iterate — re-run the generator as the target module evolves; always re-generate from source of truth.
5. Edge Cases & Error Handling
- Missing file — a
FileNotFoundErroris reported; the demo module is written as a fallback. - Malformed default values —
_safe_reprusesast.literal_eval; on failure it falls back toast.unparsefor display, and the generated code is marked with the original repr. - Unannotated parameters — typed as
Anyin the generated signature; works at runtime but is not inspectable by the MCP client's schema. *args/**kwargs— silently excluded. The generated tool will accept keyword-only calls from the client but not positional varargs.- Decorators — not parsed in detail;
@staticmethodand@classmethodare recognized. Custom decorators do not affect the generated wrapper (the tool receives the same args). - Nested classes — only module-level functions and class methods are reflected; class names are ignored.
async deforiginal functions — the wrapper usesawait, making it async-compatible withFastMCP's async support.- Self-import — the generated file uses a lazy
from <module> import ...inside each function body, avoiding circular import errors when the MCP server is loaded at startup. - Re-generation — running the generator twice overwrites the output cleanly; the file is always overwritten, not appended to.