py2mcp#
py2mcp: Quick MCP server creation from Python functions.
Pass ordinary Python functions and get back a Model Context Protocol (MCP)
server, built on FastMCP, with each function registered as a tool. The
mk_mcp_* builders return a server object and leave running it to you:
mcp.run() for stdio, or py2mcp.serve and py2mcp.http for a
packaged stdio launcher and a Streamable-HTTP (optionally OAuth 2.1) server.
Main entry points:
mk_mcp_server: functions in,FastMCPserver outmk_mcp_from_refs: the same from'module:function'stringsmk_mcp_from_store: list/get/set/delete tools over anyMutableMappingmk_input_trans: per-argument conversion of tool inputsserve_stdioandserve_http: build from refs and runUsageLogger: one record per tool call, to a sink (py2mcp usagereads it)
>>> from py2mcp import mk_mcp_server
>>> def add(a: int, b: int) -> int:
... '''Add two numbers'''
... return a + b
>>> mcp = mk_mcp_server([add])
>>> mcp.name
'py2mcp Server'
>>> # mcp.run() # Start the server over stdio
Functions
|
Create an MCP server from Python functions. |
|
Create an MCP server from a MutableMapping with CRUD operations. |
|
Create an MCP server from |
|
Create an input transformation function from name->func mappings. |
|
Resolve a |
|
Prefilled "Add custom connector" link for claude.ai. |
|
Markdown link that installs an MCP server as a claude.ai connector. |
|
Build an MCP server from |
|
Merge a config file and explicit |
|
Load a server config JSON of the form |
|
Build a Streamable-HTTP ASGI app from |
|
Build and run a Streamable-HTTP MCP server (blocking) via FastMCP/uvicorn. |
|
Build a FastMCP resource-server auth provider from an auth-config dict. |
|
A sink that does |
Classes
|
FastMCP middleware: one record per tool call (and per handshake) to |
|
Append records to one JSON-lines file per UTC day; prune old days. |
- class py2mcp.JsonlSink(root=None, *, retention_days=90, server='server', strict=False)[source]#
Bases:
objectAppend records to one JSON-lines file per UTC day; prune old days.
Files are
<root>/<YYYY-MM-DD>.jsonl. Files older thanretention_days(0keeps today only,Nonekeeps everything) are pruned at construction and on the first write of each new day. Each write is oneO_APPENDwrite of one line, so concurrent workers on the same host interleave whole lines.rootdefaults todefault_usage_dir(); it must be one directory per server, since the day files carry no server name and pruning covers the whole directory.The sink is built to be attached to a live server, so by default it does not raise: a directory that cannot be created at construction is reported via
loggingand retried on every write (strict=Trueraises instead), and a directory removed at runtime is recreated on the next write.>>> import tempfile >>> sink = JsonlSink(tempfile.mkdtemp(), retention_days=None) >>> sink({'id': 'a', 'ts': '2026-10-02T10:00:00+00:00', 'event': 'tool_call'}) >>> sorted(p.name for p in sink.root.iterdir()) ['2026-10-02.jsonl']
- class py2mcp.UsageLogger(sink=None, *, name='server', version=None, caller=<function token_caller>, outcome=<function default_outcome>, include_args=True, redact=(), max_args_chars=4000, max_error_chars=500, handshakes=True)[source]#
Bases:
MiddlewareFastMCP middleware: one record per tool call (and per handshake) to
sink.- Parameters:
sink (
Optional[Callable[[dict],Any]]) –record -> None(sync orasync). Default: aJsonlSinkunderdefault_usage_dir()forname. Any list’sappendworks for tests;mapping_sink()adapts a store.name (
str) – The connector/server name written into every record (and the default sink’s directory).version (
Optional[str]) – The connector version written into every record, if known.caller (
Callable[[],Optional[str]]) –() -> str | Noneresolving the caller for the in-flight request. Defaulttoken_caller().outcome (
Callable[[str,Any],Union[str,Mapping[str,Any]]]) –(tool_name, result) -> label | mappingclassifying a result that did not raise. Defaultdefault_outcome(). A mapping is merged into the record (it must carry"outcome"), so a connector that knows its own “no match” shape can also set the trueresult_count.include_args (
bool) – Record the tool arguments at all.Trueby default — the arguments are the point of the log — but they are the users’ questions, so a connector that must not keep them setsFalse.redact (
Collection[str]) – Top-level argument names whose values are replaced withREDACTED(nested keys are not inspected; pre-shape the arguments withinput_transor drop them withinclude_args).max_args_chars (
Optional[int]) – Cap on the serialized arguments stored per record (Nonefor no cap).max_error_chars (
int) – Cap on the error text stored per record.handshakes (
bool) – Also recordinitializerequests (asevent: "initialize"with the client’s name/version, andoutcomeerrorif the handshake failed), so a connector that is enabled can be told from one that is used.
Error text (an exception message, or an
is_errorresult’s text) is recorded only wheninclude_argsis true andredactis empty, because it routinely echoes the arguments; otherwise onlyerror_typeis kept. The logger never raises into the tool call: a failing sink, caller or classifier is reported vialogging(loggerpy2mcp.usage) and the call proceeds. A cancelled call is recorded ascancelledwith no text; other ``BaseException``s (shutdown) pass through unrecorded.>>> records = [] >>> logger = UsageLogger(records.append, name='demo', redact=('token',)) >>> logger.name, logger.redact, logger.records_error_text ('demo', frozenset({'token'}), False)
- py2mcp.claude_install_link(name, mcp_url, *, admin=False)[source]#
Prefilled “Add custom connector” link for claude.ai.
There is no true one-click install for an unlisted MCP server (listing requires Anthropic review), but this link opens the add-connector modal with the name and URL already filled in, so the user only has to confirm — which beats “go to Settings, find Connectors, paste this long URL”.
- Parameters:
name (
str) – Connector name to prefill (what the user will see in their list).mcp_url (
str) – Full URL of the MCP endpoint, e.g.https://host/api/x/mcp.admin (
bool) – Target the org-wide install page instead of the per-user one. Use it when an admin is rolling the connector out to a whole workspace; the default (False) is the personal install.
- Return type:
- Returns:
The claude.ai URL, safe to paste into a README or a chat message.
>>> claude_install_link('snout', 'https://example.com/api/snout_mcp/mcp') 'https://claude.ai/customize/connectors?modal=add-custom-connector&connectorName=snout&connectorUrl=https%3A%2F%2Fexample.com%2Fapi%2Fsnout_mcp%2Fmcp'
Names and URLs are percent-encoded, so spaces (and
&) can’t break the query string:>>> claude_install_link('my server', 'https://x.io/mcp', admin=True) 'https://claude.ai/admin-settings/connectors?modal=add-custom-connector&connectorName=my%20server&connectorUrl=https%3A%2F%2Fx.io%2Fmcp'
Note that the link is a convenience, not an access grant: if the server is an OAuth resource server with an allowlist, a user who isn’t on it can follow the link, complete the flow, and still be refused. Custom connectors are also a paid-plan feature, so the link goes nowhere for a Free-plan user.
- py2mcp.import_object(ref)[source]#
Resolve a
'module.path:attr'(preferred) or'module.path.attr'reference.Useful for building MCP servers from configuration strings (e.g. tool references declared in a file), so callers don’t reimplement the
importlibdance. With a colon, everything before it is the module and everything after it an attribute path; without one, the last dot splits module from attribute, so'pkg.mod.Class.method'cannot be reached in the dotted form (use'pkg.mod:Class.method').- Parameters:
ref (
str) – The reference string; themodule:attrform is preferred.- Return type:
- Returns:
The object the reference names, after importing its module.
- Raises:
ValueError – The reference has no module part or no attribute part.
ImportError – The module part does not import (
ModuleNotFoundErrorwhen the module does not exist; a plainImportErrorif it exists but fails while importing).AttributeError – The attribute path does not exist on the module.
Examples
>>> import_object('json:dumps') <function dumps at ...> >>> import_object('os.path.join') <function join at ...> >>> import_object('no-separator') Traceback (most recent call last): ... ValueError: Invalid object reference 'no-separator'; expected 'module:attr' or 'module.path.attr'.
See also
py2mcp.mk_mcp_from_refs(): build a server from such references.
- py2mcp.load_server_config(path)[source]#
Load a server config JSON of the form
{"name": str, "refs": [str, ...]}.Only the shape is checked here; the refs are resolved later, when the server is built. An actionable error beats a server that starts with no tools.
- Parameters:
- Return type:
- Returns:
The parsed JSON object, untouched, with at least a
refslist.- Raises:
ValueError – The file is not valid JSON, or is not a JSON object carrying a
refslist.OSError – The file cannot be read.
Examples
>>> import json, tempfile >>> from pathlib import Path >>> path = Path(tempfile.mkdtemp()) / 'py2mcp_config.json' >>> _ = path.write_text(json.dumps({'name': 'My Tools', 'refs': ['os.path:basename']})) >>> load_server_config(path) {'name': 'My Tools', 'refs': ['os.path:basename']} >>> _ = path.write_text(json.dumps({'name': 'no refs here'})) >>> load_server_config(path) Traceback (most recent call last): ... ValueError: py2mcp server config '...py2mcp_config.json' must be a JSON object with a "refs" list, ...
See also
resolve_server_config(): merge the config with command-line refs.
- py2mcp.mapping_sink(store, *, key=<function _default_key>)[source]#
A sink that does
store[key(record)] = record— anyMutableMapping.The default key is
"<YYYY-MM-DD>/<id>.json", so adolfile store lays records out one per file by day, and “last 30 days” is a key-prefix scan. The write happens on the event loop, so this is for a local store; a remote one (S3, a database) belongs behind anasyncsink or a batching one.>>> store = {} >>> sink = mapping_sink(store) >>> sink({'id': 'abc', 'ts': '2026-10-02T10:00:00+00:00'}) >>> list(store) ['2026-10-02/abc.json']
- py2mcp.markdown_install_badge(name, mcp_url, *, admin=False)[source]#
Markdown link that installs an MCP server as a claude.ai connector.
The dominant use of
claude_install_link()is pasting one into a README, so this saves writing the same link syntax around it. Arguments are those ofclaude_install_link().>>> markdown_install_badge('snout', 'https://x.io/mcp') '[Add snout to Claude](https://claude.ai/customize/connectors?modal=add-custom-connector&connectorName=snout&connectorUrl=https%3A%2F%2Fx.io%2Fmcp)'
Brackets and backslashes in the name are escaped in the link text, so a name like
tools [beta]can’t end the link text early and break the link (the URL itself is percent-encoded already):>>> print(markdown_install_badge('tools [beta]', 'https://x.io/mcp').split('](')[0]) [Add tools \[beta\] to Claude
- Return type:
- py2mcp.mk_auth_provider(auth)[source]#
Build a FastMCP resource-server auth provider from an auth-config dict.
authisNone/falsy (no auth) or a dict with atypekey:type='jwt'(default) — validate JWTs issued by a managed IdP. Keys:jwks_uriorpublic_key— where to get the IdP’s signing key(s).issuer— the IdP issuer URL (the token’siss).audience(required) — this server’s resource id (the token’saud). RFC 8707 audience binding is mandatory: it stops a token minted for another service being replayed here (the confused-deputy defense), so this helper refuses to build a verifier that would skip it.authorization_servers(or a singleissuer) — IdP issuer URL(s) advertised in the RFC 9728 protected-resource metadata.base_url— this server’s public base URL.required_scopes(optional) — scopes every request must carry.
Building the provider performs no network I/O (key fetching is lazy, on the first request), so this is safe to call at scaffold/import time.
- Parameters:
auth (
Optional[dict]) – The auth-config dict described above, orNone/{}for no authentication.- Return type:
- Returns:
A
RemoteAuthProvider(a resource server), orNonewhenauthis falsy.- Raises:
ValueError –
authis not a dict, itstypeis not supported, or a required key (jwks_uri/public_key,base_url,audience,authorization_servers/issuer) is missing.
Examples
>>> mk_auth_provider(None) is None True >>> provider = mk_auth_provider({ ... 'type': 'jwt', ... 'jwks_uri': 'https://idp.example.com/.well-known/jwks.json', ... 'issuer': 'https://idp.example.com', ... 'audience': 'https://conn.example.com/mcp', ... 'base_url': 'https://conn.example.com', ... }) >>> type(provider).__name__ 'RemoteAuthProvider' >>> provider.authorization_servers ['https://idp.example.com']
Leaving out the audience is refused rather than silently unchecked:
>>> mk_auth_provider({'type': 'jwt', 'jwks_uri': 'https://idp.example.com/jwks', ... 'base_url': 'https://conn.example.com'}) Traceback (most recent call last): ... ValueError: jwt auth needs 'audience' (this server's resource id). ...
See also
mk_http_app(): where the provider is attached to a server.
- py2mcp.mk_http_app(refs, *, name='py2mcp Server', auth=None, input_trans=None, transport='streamable-http', path=None, stateless_http=None, middleware=None, instructions=None, prompts=None, resources=None, icons=None, annotations=None)[source]#
Build a Streamable-HTTP ASGI app from
refs(+ optional OAuth).Returns the ASGI application (a Starlette app), so any ASGI server can run it:
# server/app.py from py2mcp.http import mk_http_app app = mk_http_app(['mypkg.tools:summarize'], name='My Connector', auth=AUTH) # then: uvicorn server.app:app --host 0.0.0.0 --port 8000
Builds the app with no network I/O.
- Parameters:
refs (
Iterable[str]) –'module:function'references, one per tool.name (
str) – Server name.auth (
Optional[dict]) – Resolved bymk_auth_provider()(None→ no auth; a remote connector should always set it).input_trans (
Optional[Callable[[dict],dict]]) – Forwarded topy2mcp.mk_mcp_from_refs().transport (
str) – The FastMCP HTTP transport;DFLT_TRANSPORTis Streamable HTTP.path (
Optional[str]) – URL path the MCP endpoint is mounted at;Nonekeeps FastMCP’s default (/mcp).stateless_http (
Optional[bool]) –Trueis recommended behind a load balancer (MCP sessions are stateful, so default in-memory sessions break across replicas — go stateless or externalize session state).Nonekeeps FastMCP’s default.middleware (
Optional[Any]) – A single FastMCP middleware or a list, attached for cross-cutting concerns — metering, logging, rate limiting. Becauseauthruns first, it can read the authenticated caller viafastmcp.server.dependencies.get_access_token().instructions (
Optional[str]) – The server’s model-facing description (surfaced to the connecting client/model).prompts (
Union[Callable,Iterable[Callable],None]) – Forwarded topy2mcp.mk_mcp_from_refs().resources (
Optional[Mapping[str,Callable]]) – Forwarded topy2mcp.mk_mcp_from_refs().icons (
Optional[Any]) – Forwarded topy2mcp.mk_mcp_from_refs().annotations (
Optional[Any]) – Forwarded topy2mcp.mk_mcp_from_refs()(per-tool MCP hints).
- Return type:
- Returns:
The Starlette ASGI application.
- Raises:
ValueError –
authis malformed (seemk_auth_provider()) or a reference cannot be parsed (seepy2mcp.mk_mcp_from_refs()).
Examples
>>> app = mk_http_app(['os.path:basename'], name='Paths') >>> callable(app) True >>> [route.path for route in app.routes] ['/mcp']
With OAuth and a custom mount path, the RFC 9728 protected-resource metadata route is added next to the endpoint:
>>> AUTH = { ... 'type': 'jwt', ... 'jwks_uri': 'https://idp.example.com/.well-known/jwks.json', ... 'issuer': 'https://idp.example.com', ... 'audience': 'https://conn.example.com/mcp', ... 'base_url': 'https://conn.example.com', ... } >>> app = mk_http_app(['os.path:basename'], name='Paths', auth=AUTH, path='/api/mcp') >>> [route.path for route in app.routes] ['/.well-known/oauth-protected-resource/api/mcp', '/api/mcp']
See also
serve_http(): build and run in-process instead of returning the app.py2mcp.serve.serve_stdio(): the local stdio counterpart.
- py2mcp.mk_input_trans(name_func_relationships=None)[source]#
Create an input transformation function from name->func mappings.
The returned callable takes a tool call’s keyword arguments as a dict and returns a new dict in which each named argument has been passed through its converter; arguments with no converter are copied through unchanged. Pass it as
input_transtopy2mcp.mk_mcp_server()and friends.- Parameters:
name_func_relationships (
Optional[Mapping]) – Which converter applies to which argument, as{name: func}, or reversed as{func: name}or{func: [name, ...]}when one converter serves several arguments.Nonegives a transformation that only copies the dict.- Return type:
- Returns:
A function from a kwargs dict to a new kwargs dict.
- Raises:
ValueError – The same argument name is mapped more than once.
Examples
>>> def to_int(x): return int(x) >>> trans = mk_input_trans({'x': to_int}) >>> trans({'x': '42', 'y': 'hello'}) {'x': 42, 'y': 'hello'}
One converter for several arguments, written the reversed way:
>>> trans = mk_input_trans({to_int: ['x', 'y']}) >>> trans({'x': '1', 'y': '2', 'z': '3'}) {'x': 1, 'y': 2, 'z': '3'}
No mapping means no conversion:
>>> mk_input_trans()({'x': '1'}) {'x': '1'}
See also
py2mcp.mk_mcp_server(): where the returned callable is applied.
- py2mcp.mk_mcp_from_refs(refs, *, name='py2mcp Server', input_trans=None, auth=None, middleware=None, instructions=None, prompts=None, resources=None, icons=None, annotations=None)[source]#
Create an MCP server from
'module:function'reference strings.Resolves each reference to a callable via
py2mcp.util.import_object()and delegates tomk_mcp_server(). One call from config strings to a runnable server — what tools that read tool references from a file (e.g.coact’smcpbackend) need.- Parameters:
refs (
Iterable[str]) –'module.path:attr'(or'module.path.attr') strings, one per tool. Every module is imported when this is called.name (
str) – Name of the MCP server.input_trans (
Optional[Callable[[dict],dict]]) – Forwarded tomk_mcp_server().auth (
Optional[Any]) – Forwarded tomk_mcp_server(); the remote/HTTP path attaches its OAuth provider here.middleware (
Optional[Any]) – Forwarded tomk_mcp_server().instructions (
Optional[str]) – Forwarded tomk_mcp_server()as the server’s model-facing description.prompts (
Union[Callable,Iterable[Callable],None]) – Forwarded tomk_mcp_server().resources (
Optional[Mapping[str,Callable]]) – Forwarded tomk_mcp_server().icons (
Optional[Any]) – Forwarded tomk_mcp_server().annotations (
Optional[Any]) – Forwarded tomk_mcp_server().
- Return type:
FastMCP- Returns:
A FastMCP server with one tool per reference, each named after the resolved function.
- Raises:
ValueError – A reference has no
moduleorattrpart (seepy2mcp.util.import_object(), whose import errors propagate too).
Examples
>>> mcp = mk_mcp_from_refs(['os.path:basename', 'os.path:dirname'], name='Paths') >>> mcp.name 'Paths'
The tools are the resolved functions:
>>> import asyncio >>> sorted(tool.name for tool in asyncio.run(mcp.list_tools())) ['basename', 'dirname'] >>> asyncio.run(mcp.call_tool('basename', {'p': '/a/b/c.txt'})).content[0].text 'c.txt'
See also
py2mcp.serve.serve_stdio(): build from refs and run over stdio.py2mcp.http.mk_http_app(): build from refs as an ASGI app.
- py2mcp.mk_mcp_from_store(store, *, name='item', plural='', server_name=None, middleware=None, instructions=None, prompts=None, resources=None, icons=None, annotations=None)[source]#
Create an MCP server from a MutableMapping with CRUD operations.
Generates four tools over the store,
list_<plural>,get_<name>,set_<name>anddelete_<name>, so any key-value store (a dict, adolstore, a database wrapper) is one call away from being MCP tools. The store is used live: a tool call reads or writes the mapping you passed.- Parameters:
store (
MutableMapping[Any,Any]) – A MutableMapping to expose via MCP.name (
str) – Singular name for items (e.g., ‘project’, ‘user’); used in the tool names.plural (
str) – Plural form used by the list tool (defaults to name + ‘s’).server_name (
Optional[str]) – Name of the MCP server (defaults to “{name} Store”).middleware (
Optional[Any]) – Optional FastMCP middleware (a single middleware or an iterable), forwarded tomk_mcp_server()— wraps every generated CRUD tool call, e.g. to meter or audit store reads and mutations.instructions (
Optional[str]) – Optional natural-language server description, forwarded tomk_mcp_server()as the server’s model-facinginstructions.prompts (
Union[Callable,Iterable[Callable],None]) – Forwarded tomk_mcp_server().resources (
Optional[Mapping[str,Callable]]) – Forwarded tomk_mcp_server().icons (
Optional[Any]) – Forwarded tomk_mcp_server().annotations (
Optional[Any]) – Forwarded tomk_mcp_server(); keys are the generated tool names (e.g.delete_project).
- Return type:
FastMCP- Returns:
A FastMCP server with the four CRUD tools.
Examples
>>> import asyncio >>> projects = {'p1': {'name': 'Project 1'}, 'p2': {'name': 'Project 2'}} >>> mcp = mk_mcp_from_store(projects, name='project') >>> mcp.name 'project Store' >>> sorted(tool.name for tool in asyncio.run(mcp.list_tools())) ['delete_project', 'get_project', 'list_projects', 'set_project'] >>> asyncio.run(mcp.call_tool('get_project', {'key': 'p1'})).structured_content {'result': {'name': 'Project 1'}}
An irregular plural and an explicit server name:
>>> mcp = mk_mcp_from_store({}, name='entry', plural='entries', server_name='Ledger') >>> mcp.name 'Ledger' >>> sorted(tool.name for tool in asyncio.run(mcp.list_tools())) ['delete_entry', 'get_entry', 'list_entries', 'set_entry']
See also
py2mcp.util.store_to_funcs(): the CRUD functions without a server.mk_mcp_server(): expose your own functions instead.
- py2mcp.mk_mcp_server(funcs, *, name='py2mcp Server', input_trans=None, auth=None, middleware=None, instructions=None, prompts=None, resources=None, icons=None, annotations=None)[source]#
Create an MCP server from Python functions.
This is the main entry point for py2mcp. Pass one or more functions, and get back a FastMCP server ready to run. Each function becomes one tool, named after the function, with its signature and docstring as the tool’s schema and description.
- Parameters:
funcs (
Union[Callable,Iterable[Callable]]) – A function or iterable of functions to expose as MCP tools.name (
str) – Name of the MCP server.input_trans (
Optional[Callable[[dict],dict]]) – Called with the dict of keyword arguments of every tool call; the dict it returns is what the function receives. Build one withpy2mcp.mk_input_trans().Nonepasses arguments through untouched.auth (
Optional[Any]) – Optionalfastmcp.server.authprovider attached at construction — used by the remote (HTTP) path for OAuth 2.1 (seepy2mcp.http).None(the default) leaves the server unauthenticated, which is correct for the local stdio path.middleware (
Optional[Any]) – Optional FastMCP middleware (a single middleware or a list), attached at construction, for cross-cutting concerns that must wrap every tool call — usage metering, cost logging, audit, rate limiting. Preferred over decorating each tool: you can’t forget to wrap one (a missed paid tool means untracked cost). On the remote pathauthruns first, so a middleware can read the authenticated caller viafastmcp.server.dependencies.get_access_token().instructions (
Optional[str]) – Optional natural-language description of the server, surfaced to the client/model as the server’sinstructions— a good place to explain what the tools do and the intended workflow.None(default) leaves it unset.prompts (
Union[Callable,Iterable[Callable],None]) – Optional callable or iterable of callables to register as MCP prompts (via FastMCP’s@mcp.prompt), so a prompts-and-tools server can be built declaratively in one call instead of reaching past the builder to register prompts by hand on the returned server.resources (
Optional[Mapping[str,Callable]]) – Optional{uri: callable}mapping to register as MCP resources (via FastMCP’s@mcp.resource(uri)) — the callable is invoked to produce the resource’s content when a client readsuri.icons (
Optional[Any]) – Optional server icon(s), sent in the MCPserverInfoso a client (e.g. claude.ai’s connector list) can show one: anmcp.types.Icon, a dict of its fields, a baresrc(URL ordata:URI), or a list of those.None(default) sends none.annotations (
Optional[Any]) – Optional per-tool MCP annotations (readOnlyHint,destructiveHint,idempotentHint,openWorldHint), so a host can gate tools by them instead of treating all as destructive. Either a{tool_name: annotations}mapping (tools absent from it get none), or a callablefunc -> annotations | Noneevaluated per function (e.g. to derive them from your own side-effect metadata). Each value is anmcp.types.ToolAnnotationsor a dict of its fields.None(default) sets none.
- Return type:
FastMCP- Returns:
A FastMCP server instance ready to run, with one tool per function.
Examples
>>> def add(a: int, b: int) -> int: ... '''Add two numbers''' ... return a + b >>> mcp = mk_mcp_server(add) >>> mcp.name 'py2mcp Server'
Several functions, a server name, and a look at the registered tools:
>>> import asyncio >>> def greet(name: str) -> str: ... return f"Hello, {name}!" >>> mcp = mk_mcp_server([add, greet], name="Math & Greetings") >>> mcp.name 'Math & Greetings' >>> sorted(tool.name for tool in asyncio.run(mcp.list_tools())) ['add', 'greet']
Calling a tool the way an MCP client would:
>>> result = asyncio.run(mcp.call_tool('add', {'a': 2, 'b': 3})) >>> result.structured_content {'result': 5}
Prompts and resources, declared alongside the tools:
>>> def summarize_request(topic: str) -> str: ... return f"Summarize the latest on {topic}." >>> def schema() -> dict: ... return {"type": "object"} >>> mcp = mk_mcp_server( ... add, ... prompts=summarize_request, ... resources={"schema://analysis": schema}, ... ) >>> sorted(p.name for p in asyncio.run(mcp.list_prompts())) ['summarize_request'] >>> [str(r.uri) for r in asyncio.run(mcp.list_resources())] ['schema://analysis']
Annotations, by tool name (or a callable
func -> annotations):>>> mcp = mk_mcp_server([add, greet], annotations={'add': {'readOnlyHint': True}}) >>> tools = {t.name: t for t in asyncio.run(mcp.list_tools())} >>> tools['add'].annotations.readOnlyHint, tools['greet'].annotations (True, None)
See also
mk_mcp_from_refs(): the same from'module:function'strings.mk_mcp_from_store(): CRUD tools generated from a mapping.py2mcp.mk_input_trans(): build aninput_transfrom per-argument converters.
- py2mcp.resolve_server_config(*, config=None, refs=(), name=None)[source]#
Merge a config file and explicit
refs/nameinto(refs, name).Refs from
--refare appended after any from the config file; an explicitnamewins over the config’s. Pure (no I/O beyond readingconfig), so it is unit-testable without standing up a server.- Parameters:
config (
Union[str,Path,None]) – Path of a JSON config as read byload_server_config(), orNonefor no config file.refs (
Iterable[str]) – Extra'module:function'references, appended after the config’s.name (
Optional[str]) – Server name; overrides the config’sname. When neither is given,DFLT_SERVER_NAME.
- Return type:
- Returns:
The merged list of references and the server name.
- Raises:
ValueError – Neither the config nor
refssupplies any reference, or the config file is malformed (seeload_server_config()).
Examples
>>> resolve_server_config(refs=['os.path:basename'], name='Paths') (['os.path:basename'], 'Paths')
With a config file, its refs come first and its name is the fallback:
>>> import json, tempfile >>> from pathlib import Path >>> path = Path(tempfile.mkdtemp()) / 'py2mcp_config.json' >>> _ = path.write_text(json.dumps({'name': 'My Tools', 'refs': ['os.path:basename']})) >>> resolve_server_config(config=path, refs=['os.path:dirname']) (['os.path:basename', 'os.path:dirname'], 'My Tools') >>> resolve_server_config(config=path, name='Override') (['os.path:basename'], 'Override')
- py2mcp.serve_http(refs, *, name='py2mcp Server', host='127.0.0.1', port=8000, auth=None, input_trans=None, transport='streamable-http', stateless_http=None, middleware=None, instructions=None, prompts=None, resources=None, icons=None, annotations=None)[source]#
Build and run a Streamable-HTTP MCP server (blocking) via FastMCP/uvicorn.
For a self-hosted process. Binds
127.0.0.1by default — expose a public interface only behind a TLS-terminating reverse proxy (a remote connector must be reachable over public HTTPS, and binding locally is the spec’s DNS-rebinding-safe default).authis resolved bymk_auth_provider();middleware(a single FastMCP middleware or a list) is attached as inmk_http_app();instructionssets the server’s model-facing description.prompts/resourcesare forwarded topy2mcp.mk_mcp_from_refs().- Return type:
- py2mcp.serve_stdio(refs, *, name='py2mcp Server', input_trans=None, middleware=None, instructions=None, prompts=None, resources=None)[source]#
Build an MCP server from
'module:function'refs and run it over stdio.Blocks, serving the MCP protocol on stdin/stdout until the host disconnects. Thin wrapper over
py2mcp.mk_mcp_from_refs()+FastMCP.runso that packaged integrations have one command to launch. No example here: the call does not return while the server runs.- Parameters:
refs (
Iterable[str]) –'module:function'references, one per tool.name (
str) – Server name.input_trans (
Optional[Callable[[dict],dict]]) – Forwarded topy2mcp.mk_mcp_from_refs().middleware (
Optional[Any]) – A single FastMCP middleware or a list, forwarded for cross-cutting concerns; logging/metering is as useful on the local stdio path as on the remote one.instructions (
Optional[str]) – The server’s model-facing description.prompts (
Union[Callable,Iterable[Callable],None]) – Forwarded topy2mcp.mk_mcp_from_refs().resources (
Optional[Mapping[str,Callable]]) – Forwarded topy2mcp.mk_mcp_from_refs().
- Return type:
See also
py2mcp.http.serve_http(): the same over Streamable HTTP.main(): the command line that calls this.
Modules
Private helpers shared by the builders in |
|
Serve a py2mcp |
|
Build a |
|
Serve a py2mcp |
|
Input transformation for py2mcp tools: convert arguments before a function runs. |
|
Per-call usage logging for a py2mcp / FastMCP server: who called which tool, with what, and what came back. |
|
Resolve object references and turn a mapping into CRUD functions. |