MCP

Opt-in MCP (Model Context Protocol) tool surface over a duho CLI — a static _subcommands_ class tree, or a full duho.app()-built tree (class AND module commands) — via describe_tools/call_tool, and a stdio server runnable as python -m duho.mcp <app>. import duho.mcp to use it; core duho never imports this module, and neither does a normal duho.main/duho.app run — the <PREFIX>MCP/<NAME>_MCP launch trigger (default on; opt out with _mcp_ = False/app(mcp=False)) and the opt-in McpCmd/_mcp_command_/app(mcp_command=...) subcommand (registered under either duho.main or duho.app — _mcp_command_ reads the same way from both) both import it lazily, only once actually triggered. A non-root command opts OUT of the tool surface entirely with its own _mcp_ = False (a module command via a module-level _mcp_ = False) — excluded together with its whole subtree, refused the same way an unknown tool name is.

duho.mcp

MCP tool surface over a duho CLI (opt-in, standalone module).

Exposes the same Cmd/Cli classes that back a duho CLI as MCP tools (Model Context Protocol -- https://modelcontextprotocol.io), with zero redeclaration. duho already does the hard part: :func:duho._introspect.get_clsargs / :class:~duho._introspect.ClsArgDeclaration plus each field's :class:~duho.args.ArgumentBuilder produce, per field, a type/default/docstring -- exactly the raw material an MCP tool's inputSchema needs. argparse is just one frontend rendered from those declarations; this module is a second frontend over the same data, dispatching through :func:duho.run_command verbatim.

Opt-in / off the core surface, like duho.runpath/duho.fanout/ duho.scaffold: core duho never imports this module, and it is deliberately not on the top-level duho.* surface. Activate it with import duho.mcp or python -m duho.mcp <app>.

Zero-dep stdlib JSON-RPC over stdio -- no MCP SDK dependency. json (and importlib.metadata, transitively reachable via pkgutil.resolve_name's import machinery for the <app> CLI arg) are imported lazily, function-local, never at module top -- so import duho.mcp alone never pays their cost, mirroring the zero-eager-import contract import duho already keeps (see tests/test_config_json.py, tests/test_entry_points.py).

Three layers, thin glue between them:

• :func:input_schema_for_command / :func:json_schema_for_field -- step 1, the type -> JSON Schema emitter. Standalone and parser-free: it reads a command class's own field declarations directly (cls._getargs_() + duho._introspect.get_clsargs(cls)), not a built argparse parser.
• :func:describe_tools -- step 2. Walks the class's own STATIC _subcommands_ tree (built once per root class and cached), reusing the same alias-dedup-by-identity pattern as duho.agenthelp.describe_parser, so every Cmd in that tree becomes one MCP tool, namespaced parent.child when nested -- except a namespace node whose own subcommand is mandatory (_subcommands_ always registers a required argparse subparsers action): such a node can never itself dispatch successfully, so it is not listed as a tool at all; its own fields are still merged into every one of its descendants' schemas, and it is still reachable through them. Two kinds of tree are served: a Cmd/Cli class's static _subcommands_ tree (what :func:serve and python -m duho.mcp <app> take, so <app> there must be a Cmd/Cli subclass), and a full duho.app() tree -- class AND module commands from source=/commands=/CMDS_PATH -- served through the <NAME>_MCP environment trigger, an mcp_command subcommand, or :func:serve_running_app.
• :func:call_tool -- step 3. Resolves a tool name back to its node in that same cached tree, synthesizes an argv for its FULL path (its own ancestors' fields, in order, with the subcommand name token between each level, then its own fields), and dispatches through the tree's single shared ROOT parser -- exactly the parser duho.main/duho.parse would build and parse, so env/config layering (Meta(env=...), _config_), root globals, _passthrough_, and LoggingArgs' own -v/-q/ --loglevel verbosity setup all work over MCP exactly as they do on the CLI. Maps the result per the return convention (see below).

Return convention: during a call, stdout (and stderr) are captured. A command returning None/0 -> a success result whose one text content block is the captured stdout (empty string allowed). A non-zero int -> isError: true, text = captured stdout + a trailing "exit code: N" line ("exit code: N (meaning)" when the root's own _exit_codes_ declares N). A duho.Result with text or value -> captured stdout then that text (else the rendered value), isError from result.is_error. A duho.CommandError, or an exception the root's _errors_ maps -> isError: true with captured stdout, stderr and the message, no Type: prefix. A JSON-serialisable object/list return -> passed through as one text block holding its JSON dump. A command that calls sys.exit(...)/raises SystemExit at RUN TIME (not during argument parsing) is mapped the same way as a returned exit code, from exc.code: None/0 is success, an int is isError with an "exit code: N" line, anything else (e.g. sys.exit("message")) is isError with that message -- call_tool never lets a command's own SystemExit escape and end the whole server process. Any OTHER raised exception during dispatch is mapped to isError: true with the exception's type: message text. KeyboardInterrupt (and any other non-SystemExit BaseException) is deliberately NOT caught, so it still stops the server.

Output that is not text: sys.stdout has a buffer. Output that is valid UTF-8 is the text above; from the first bytes that are not, all of it goes to the server's output target (descriptor 3, or what the <NAME>_MCP_OUTPUT variable names: a descriptor number or a file or pipe name), or the call fails when there is none. A command's _mcp_output_ ("auto", "stream" or "result") changes that; see :func:call_tool.

Malformed requests are a different kind of failure than a broken command: :func:call_tool raises :class:UnknownToolError for a tool name that does not resolve (or names a namespace node -- see above), and :class:InvalidArgumentsError when arguments is not a JSON object, fails the tool's own inputSchema (an unknown property, a missing required one, a value whose JSON type/enum does not match, a numeric value outside its minimum/maximum, or a collection over its maxItems/ maxProperties), or -- detected while synthesizing argv, since it depends on the built parser tree rather than the schema alone -- supplies a value that cannot be safely encoded at all (an unsafe positional, an unsafe option value, or a negative counting-flag value; see :func:call_tool's own "Dispatch-identity guard" section). All of these are :class:ValueError subclasses carrying a JSON-RPC .code (-32602, "invalid params"); :func:serve maps them to a JSON-RPC error response, never a tool result -- the request itself, not the target command, was invalid. The one exception is a session that negotiated protocol revision 2025-11-25 or later: there an :class:InvalidArgumentsError for a JSON-object arguments (a schema failure or an unsafe value) is a successful response whose result is {"content": [{"type": "text", "text": <message>}], "isError": true}, so the model can read the message and correct the call. An unknown tool, arguments that is not an object, and any request before initialize stay JSON-RPC errors. :func:call_tool raises in every case. A problem in the dispatched command itself (a raised exception, a non-zero exit, sys.exit, or an argparse usage error from a value that WAS schema-valid but the command still rejects) is still a normal tool result with isError: true.

Documented v1 limitations: a custom action=/type= field with no registered override is passed through as a plain string, verbatim; Meta(conflicts=...) exclusive groups are surfaced only as a note appended to the tool's description text (no oneOf/not JSON Schema encoding yet); a field that defaults to True and declares only short flags (no long flag) cannot be turned back to False over MCP (there is no --no-<x> form to emit) and raises rather than silently doing the wrong thing; a module command whose register() hook adds its own subparsers is listed as that one tool; the hand-made subparsers are not tools of their own: one is chosen through a property named after the subparsers' dest, and their own options are not served; a value equal to the literal string "--" is refused as a positional value and for a field with only a short flag (argparse's own -- end-of-options marker, and duho's own _passthrough_ split, make it unsafe there -- use the dedicated "--" array property instead, see :func:_input_schema_for_node); it is accepted as a long-flag option value; streaming/long-running commands are out of scope -- this is strictly one request -> one result.

__all__ = ['json_schema_for_field', 'input_schema_for_command', 'describe_tools', 'call_tool', 'serve', 'serve_running_app', 'McpCmd', 'main', 'UnknownToolError', 'InvalidArgumentsError'] module-attribute

InvalidArgumentsError

Bases: ValueError

Raised by :func:call_tool when arguments is not a JSON object, or fails the target tool's own inputSchema (an unknown property, or a value whose JSON type does not match). Maps to a JSON-RPC -32602 error response in :func:serve, never a tool result.

code = -32602 class-attribute instance-attribute

McpCmd(**kwargs)

Bases: Cmd

Serve the CLI currently being dispatched as an MCP server (stdio).

A ready-made building block: register a subclass of this under any name to add a self-serving MCP subcommand to a CLI, with zero server code of your own -- __call__ just forwards to :func:serve_running_app. duho.app's own opt-in _mcp_command_ class attribute / mcp_command= kwarg do exactly this (a dynamically named subclass), but any app may register McpCmd (or a subclass adding, say, its own additional flags) under any name it likes, via _subcommands_/commands=/self-registration -- there is nothing _mcp_command_ does that isn't equally reachable by hand.

A node whose class IS (or subclasses) McpCmd is never listed as an MCP tool, and never callable via call_tool either (see :func:_is_mcp_command_node) -- a client asking a live MCP server to recursively take over stdio again makes no sense.

Source code in src/duho/args/_argsclass.py

def __init__(self, /, **kwargs: object) -> None:
    # Seed each declared field's effective default when absent, so a direct
    # instance has the attribute surface of a parsed one (e.g. a `store_true`
    # bool). Passed kwargs win; a required field (NOT_DEFINED) stays unset.
    # Not `super().__init__(**kwargs)`: a field named `self` would collide.
    super().__init__()
    for key, value in kwargs.items():
        setattr(self, key, value)
    if "_passthrough_" not in vars(self):
        self._passthrough_ = []
    seeded: dict[str, object] = {}
    for builder in type(self)._getargs_():
        name = builder.name
        # `vars(self)`, not `hasattr`: a class-level default is `hasattr`-true
        # by inheritance, so the instance would share, and mutations would
        # leak through, the class attribute.
        if name in kwargs or name in vars(self):
            continue
        default = builder._effective_default_()
        if default is not NOT_DEFINED:
            setattr(self, name, default)
            seeded[name] = _copy.copy(default)
    # Record which fields THIS CALL passed, outside vars(self). A class that
    # is not weak-referenceable is untracked, and `duho.parse(instance)`
    # then treats every attribute as explicit.
    try:
        _key = id(self)
        _duho_explicit_instance_fields[_key] = frozenset(kwargs)
        _duho_seeded_instance_values[_key] = seeded
        _weakref.finalize(self, _duho_explicit_instance_fields.pop, _key, None)
        _weakref.finalize(self, _duho_seeded_instance_values.pop, _key, None)
    except TypeError:
        pass

transport = 'stdio' class-attribute instance-attribute

MCP transport to serve this CLI over

__call__()

Source code in src/duho/mcp/_command.py

def __call__(self) -> int:
    from ._stdio import serve_running_app

    return serve_running_app(self.transport)

UnknownToolError

Bases: ValueError

Raised by :func:call_tool for a tool name that does not resolve to a callable node in the root class's tree (including a namespace node whose own subcommand is mandatory -- see the module docstring). Maps to a JSON-RPC -32602 error response in :func:serve, never a tool result.

code = -32602 class-attribute instance-attribute

call_tool(root_cls, name, arguments, *, output_fd=None, output_var=None, output_path=None)

Dispatch one MCP tools/call against root_cls's tree.

root_cls is a Cmd/Cli class (the static _subcommands_ tree); the :class:_ServerCore form is built by duho for an app() tree, not by a caller -- see :func:describe_tools.

Resolves name to a node in the tree, raising :class:UnknownToolError for a name that is not in the tree, that names a namespace node (see :func:_is_namespace_node), or that names a node excluded via a per-command _mcp_ = False (:attr:_Node.excluded) -- the same error either way, so an excluded command's existence is never disclosed to a caller probing for it. Raises :class:InvalidArgumentsError when arguments is not a JSON object (None is treated as {}), fails the tool's own merged inputSchema (:func:_validate_arguments), or (detected while synthesizing argv, since it depends on the built parser tree rather than the JSON schema alone) supplies a value that cannot be safely encoded at all -- an unsafe positional, an unsafe option value, or a negative counting-flag value (see :func:_synthesize_argv/:func:_reject_unsafe_positional/ :func:_emit_option). All of these are request-level problems, mapped by :func:serve to a JSON-RPC error response rather than a tool result.

Otherwise: synthesizes one argv per level of the tool's ancestry chain (root first) via :func:_synthesize_argv, with the next level's own subcommand name token in between, then -- when the JSON arguments carry a "--" array -- one literal -- token followed by its items at the very end (see :func:_input_schema_for_node's synthetic property), and parses the WHOLE THING through the tree's single shared ROOT parser -- exactly as duho.main/duho.parse would for the equivalent CLI invocation, so env/config layering, root globals, _passthrough_, and (via :func:duho.args._setup_instance_logging) LoggingArgs verbosity setup all reach the dispatched command. Captures stdout AND stderr during dispatch, and replaces sys.stdin with an empty stream for its duration (a command honoring '-' = stdin must not be able to read the MCP client's next request off the real stdin).

Dispatch-identity guard (security-relevant: MCP tool arguments are LLM-controlled): an ancestor's own optional/variadic positional field can -- when a client omits it -- still absorb the LITERAL subcommand-name token this function inserts between levels, shifting a LATER token into that ancestor's own subparsers action and dispatching a DIFFERENT sibling than the one named by name (argparse itself has always allowed this; it is normally harmless on a real, human-typed CLI, but not when the argv comes from an LLM). The SAME class can also be reached from more than one place in the tree (shared between two parents, or both nested and top-level), so a class-identity check alone cannot tell a hijack from a legitimate dispatch. After a successful parse, the result is used ONLY when BOTH the intended command was actually selected (type(instance) is node.cls for a class command, or the popped _duho_module_command_ marker is node.module_command for a module command) AND the _duho_mcp_path_ tag :func:_walk_tree attaches to every subparser (via set_defaults, overwritten by the DEEPEST subparser actually reached as parsing descends) equals the tool's own chain of names exactly -- anything else (including a namespace class picked up mid-chain, or the right command reached through the WRONG chain) is treated as a dispatch failure, mapped to isError: true, and the command is NEVER run. :func:_synthesize_argv/ :func:_synthesize_argv_from_actions additionally refuse outright (before parsing, as an :class:InvalidArgumentsError) a value explicitly supplied FOR a field at any level that collides with a subcommand name/alias registered at THAT level or any ANCESTOR level.

Return convention: run_command returns 0 for a None/0 command return, an int for a non-zero return, or the raw object/list when the command returned one. Mapped here: 0 -> success, one text block of captured stdout; a non-zero int -> isError: true, captured stdout + captured stderr + a trailing "exit code: N" line; anything else -> success, one text block holding its JSON dump. A SystemExit from argument PARSING (bad/missing argument) -> isError: true with the captured stderr text. A SystemExit raised by the command's OWN run time code is mapped by :func:_systemexit_result instead of escaping and killing the server. Any OTHER raised exception during dispatch -> isError: true with the exception's type: message text. KeyboardInterrupt is not caught and still propagates.

Standard output is a stream (sys.stdout.buffer exists). Output that is valid UTF-8 is held and is the text above. At the first bytes that are not, everything held and everything later goes straight to file descriptor output_fd, and the result says how many bytes went where and at what offset, in place of the text; with output_fd None the call is an isError result instead. output_path is the file output_fd was opened on, named in the result; output_var is the variable that names the target, quoted in the error. output_path with no output_fd is a ValueError: the path opens nothing.

A command's _mcp_output_ changes this. "stream": all output, text included, goes to output_fd as written, the result says how many bytes went where, and with no output_fd the call fails before the command runs. "result": the target is not used; text is the text item and bytes that are not text follow it as one embedded base64 resource.

Source code in src/duho/mcp/_call.py

def call_tool(
    root_cls: _ty.Union[type, _ServerCore],
    name: object,
    arguments: object,
    *,
    output_fd: _ty.Optional[int] = None,
    output_var: _ty.Optional[str] = None,
    output_path: _ty.Optional[str] = None,
) -> dict:
    """Dispatch one MCP ``tools/call`` against ``root_cls``'s tree.

    ``root_cls`` is a ``Cmd``/``Cli`` class (the static ``_subcommands_``
    tree); the :class:`_ServerCore` form is built by duho for an ``app()``
    tree, not by a caller -- see :func:`describe_tools`.

    Resolves ``name`` to a node in the tree, raising
    :class:`UnknownToolError` for a name that is not in the tree, that names
    a namespace node (see :func:`_is_namespace_node`), or that names a node
    excluded via a per-command ``_mcp_ = False`` (:attr:`_Node.excluded`) --
    the same error either way, so an excluded command's existence is never
    disclosed to a caller probing for it. Raises :class:`InvalidArgumentsError` when
    ``arguments`` is not a JSON object (``None`` is treated as ``{}``), fails
    the tool's own merged ``inputSchema`` (:func:`_validate_arguments`), or
    (detected while synthesizing argv, since it depends on the built
    parser tree rather than the JSON schema alone) supplies a value that
    cannot be safely encoded at all -- an unsafe positional, an unsafe
    option value, or a negative counting-flag value (see
    :func:`_synthesize_argv`/:func:`_reject_unsafe_positional`/
    :func:`_emit_option`). All of these are request-level problems,
    mapped by :func:`serve` to a JSON-RPC error response rather than a tool
    result.

    Otherwise: synthesizes one argv per level of the tool's ancestry chain
    (root first) via :func:`_synthesize_argv`, with the next level's own
    subcommand name token in between, then -- when the JSON arguments carry
    a ``"--"`` array -- one literal ``--`` token followed by its items at
    the very end (see :func:`_input_schema_for_node`'s synthetic property),
    and parses the WHOLE THING through the tree's single shared ROOT parser
    -- exactly as ``duho.main``/``duho.parse`` would for the equivalent CLI
    invocation, so env/config layering, root globals, ``_passthrough_``, and
    (via :func:`duho.args._setup_instance_logging`) ``LoggingArgs``
    verbosity setup all reach the dispatched command. Captures stdout AND
    stderr during dispatch, and replaces ``sys.stdin`` with an empty stream
    for its duration (a command honoring '-' = stdin must not be able to
    read the MCP client's next request off the real stdin).

    **Dispatch-identity guard** (security-relevant: MCP tool arguments are
    LLM-controlled): an ancestor's own optional/variadic positional field can
    -- when a client omits it -- still absorb the LITERAL subcommand-name
    token this function inserts between levels, shifting a LATER token into
    that ancestor's own subparsers action and dispatching a DIFFERENT
    sibling than the one named by ``name`` (argparse itself has always
    allowed this; it is normally harmless on a real, human-typed CLI, but
    not when the argv comes from an LLM). The SAME class can also be reached
    from more than one place in the tree (shared between two parents, or
    both nested and top-level), so a class-identity check alone cannot tell
    a hijack from a legitimate dispatch. After a successful parse, the
    result is used ONLY when BOTH the intended command was actually selected
    (``type(instance) is node.cls`` for a class command, or the popped
    ``_duho_module_command_`` marker ``is node.module_command`` for a module
    command) AND the ``_duho_mcp_path_`` tag :func:`_walk_tree` attaches to
    every subparser (via ``set_defaults``, overwritten by the DEEPEST
    subparser actually reached as parsing descends) equals the tool's own
    chain of names exactly -- anything else (including a namespace class
    picked up mid-chain, or the right command reached through the WRONG
    chain) is treated as a dispatch failure, mapped to ``isError: true``, and
    the command is NEVER run. :func:`_synthesize_argv`/
    :func:`_synthesize_argv_from_actions` additionally refuse
    outright (before parsing, as an :class:`InvalidArgumentsError`) a value
    explicitly supplied FOR a field at any level that collides with a
    subcommand name/alias registered at THAT level or any ANCESTOR level.

    **Return convention**: ``run_command`` returns ``0`` for a
    ``None``/``0`` command return, an int for a non-zero return, or the raw
    object/list when the command returned one. Mapped here: ``0`` -> success,
    one text block of captured stdout; a non-zero int -> ``isError: true``,
    captured stdout + captured stderr + a trailing ``"exit code: N"`` line;
    anything else -> success, one text block holding its JSON dump. A
    ``SystemExit`` from argument PARSING (bad/missing argument) ->
    ``isError: true`` with the captured stderr text. A ``SystemExit`` raised
    by the command's OWN run time code is mapped by :func:`_systemexit_result`
    instead of escaping and killing the server. Any OTHER raised exception
    during dispatch -> ``isError: true`` with the exception's ``type:
    message`` text. ``KeyboardInterrupt`` is not caught and still propagates.

    Standard output is a stream (``sys.stdout.buffer`` exists). Output that is
    valid UTF-8 is held and is the text above. At the first bytes that are not,
    everything held and everything later goes straight to file descriptor
    ``output_fd``, and the result says how many bytes went where and at what
    offset, in place of the text; with ``output_fd`` ``None`` the call is an
    ``isError`` result instead. ``output_path`` is the file ``output_fd`` was
    opened on, named in the result; ``output_var`` is the variable that names
    the target, quoted in the error. ``output_path`` with no ``output_fd`` is a
    ``ValueError``: the path opens nothing.

    A command's ``_mcp_output_`` changes this. ``"stream"``: all output, text
    included, goes to ``output_fd`` as written, the result says how many bytes
    went where, and with no ``output_fd`` the call fails before the command runs.
    ``"result"``: the target is not used; text is the text item and bytes that
    are not text follow it as one embedded base64 resource.
    """
    if output_path is not None and output_fd is None:
        raise ValueError(
            "output_path names the file output_fd is open on and opens nothing; "
            "pass output_fd as well"
        )
    core = root_cls if isinstance(root_cls, _ServerCore) else _core_for_class(root_cls)
    root_parser, nodes = core.root_parser, core.nodes
    node = nodes.get(name) if isinstance(name, str) else None
    if (
        node is None
        or _is_namespace_node(node.parser)
        or _is_mcp_command_node(node)
        or node.excluded
    ):
        raise UnknownToolError("unknown tool: %r" % (name,))

    if arguments is None:
        arguments = {}
    if not isinstance(arguments, dict):
        raise InvalidArgumentsError(
            "arguments must be a JSON object, got %s" % (type(arguments).__name__,)
        )

    schema = _input_schema_for_node(node)
    _validate_arguments(schema, arguments)
    # An integral float passed validation for an integer field; argv needs
    # the integer's own spelling ("5", not "5.0").
    integer_fields = {
        key
        for key, prop in schema["properties"].items()
        if prop.get("type") == "integer"
    }
    arguments = {
        key: int(value) if key in integer_fields and isinstance(value, float) else value
        for key, value in arguments.items()
    }

    chain = node.ancestors + (node,)
    expected_path = tuple(step.own_name for step in chain)
    # A field name declared at several levels binds only at the deepest one
    # (as in `_input_schema_for_node`), so a leaf's value cannot flip an
    # ancestor's same-named flag such as a hidden `--force`.
    field_owner: dict[str, int] = {}
    for i, step in enumerate(chain):
        for fname in _step_field_names(step):
            field_owner[fname] = i
    try:
        argv: list[str] = []
        ancestor_forbidden: frozenset = frozenset()
        for i, step in enumerate(chain):
            shadowed = frozenset(
                fname for fname, owner in field_owner.items() if owner != i
            )
            argv.extend(
                _synthesize_step_argv(
                    step,
                    arguments,
                    skip=shadowed,
                    ancestor_forbidden=ancestor_forbidden,
                    pin_positionals=i + 1 < len(chain),
                )
            )
            ancestor_forbidden = ancestor_forbidden | _sibling_names(step.parser)
            if i + 1 < len(chain):
                argv.append(chain[i + 1].own_name)
        passthrough = arguments.get("--") or None
        if passthrough:
            argv.append("--")
            argv.extend(passthrough)
    except InvalidArgumentsError:
        raise
    except ValueError as exc:
        return _text_result(str(exc), is_error=True)
    # argparse expands any argv token starting with a parser's
    # `fromfile_prefix_chars` into the lines of the named file, so a value
    # must never start with one (long-flag values start with `-`, not here).
    prefixes = "".join(
        getattr(p, "fromfile_prefix_chars", None) or ""
        for p in [root_parser] + [step.parser for step in chain]
    )
    for token in argv:
        if token and token[0] in prefixes:
            raise InvalidArgumentsError(
                "value %r cannot be passed: it starts with a response-file "
                "prefix character that argparse would read as a file" % (token,)
            )

    mode = node.output
    if mode == "stream" and output_fd is None:
        return _text_result(
            "this tool sends its output to the server's output target, and this "
            "server has none: start it with file descriptor 3 open, or name a "
            "descriptor, a file or a pipe in %s" % (output_var or "<NAME>_MCP_OUTPUT"),
            is_error=True,
        )
    target = (
        None
        if output_fd is None or mode == "result"
        else _OutputTarget(output_fd, output_path)
    )
    capture = _StdoutCapture(target, mode)
    out = capture.text
    err = _io.StringIO()
    dispatched: object = None
    all_parsers = [root_parser] + [n.parser for n in nodes.values()]
    real_stderr = _sys.stderr
    traceback_token = _enter_traceback_env(core.root_cls)
    try:
        with _contextlib.redirect_stdout(out), _contextlib.redirect_stderr(err):
            old_stdin = _sys.stdin
            _sys.stdin = _io.StringIO("")
            # The parse is part of the call: a command's constructor sees "mcp" too.
            running = _compat._RUNNING_MODE.set("mcp")
            try:
                with (
                    _muted_color(all_parsers),
                    _rebound_stderr_logging(err, real_stderr),
                ):
                    try:
                        instance = root_parser.parse_args(argv)
                    except SystemExit as exc:
                        hidden = frozenset().union(
                            *(_hidden_choice_names(n.parser) for n in nodes.values())
                        )
                        message = _client_visible_parse_error(
                            err.getvalue().strip(), hidden
                        ) or ("argument error (exit code %r)" % (exc.code,))
                        return _text_result(message, is_error=True)
                    dispatched = instance
                    # Popped, as `runtime._run_app` does, so the markers never
                    # reach a command's `vars(instance)`.
                    actual_path = vars(instance).pop("_duho_mcp_path_", None)
                    dispatched_module_command = vars(instance).pop(
                        "_duho_module_command_", None
                    )
                    if node.module_command is not None:
                        identity_ok = dispatched_module_command is node.module_command
                        target: object = node.module_command
                    else:
                        identity_ok = type(instance) is node.cls
                        target = node.cls
                    if not identity_ok or actual_path != expected_path:
                        actual_desc = (
                            ".".join(actual_path)
                            if isinstance(actual_path, tuple)
                            else type(instance).__name__
                        )
                        return _text_result(
                            "tool %r did not resolve to the requested command "
                            "(dispatched %r instead); refusing to run it"
                            % (name, actual_desc),
                            is_error=True,
                        )
                    try:
                        result = core.dispatch(target, instance)
                    except SystemExit as exc:
                        taken = capture.finish(output_var)
                        if taken.error is not None:
                            return taken.error
                        return _with_blob(
                            _systemexit_result(exc, taken.text, err.getvalue()),
                            taken,
                            name,
                        )
            finally:
                _compat._RUNNING_MODE.reset(running)
                _sys.stdin = old_stdin
    except (
        Exception
    ) as exc:  # noqa: BLE001 - one broken command must not crash the server
        if capture.raw.error is not None:
            return _write_failure(output_fd, capture.raw.error)
        error_code = _error_status(exc, core.root_cls)
        if (
            error_code is None
            and isinstance(exc, _ConfigError)
            and getattr(core.root_cls, "_report_error_", None) is not None
        ):
            error_code = 2  # a config file error, as the command line reports it
        if error_code is not None:
            # An error the command meant to end with: its words, not a crash.
            reported = _run_report_hook(
                core.root_cls, exc, error_code, dispatched, capture, err, real_stderr
            )
            taken = capture.finish(output_var)
            if taken.error is not None:
                return taken.error
            parts = [taken.text, err.getvalue().strip()]
            if not reported:
                parts.append(str(exc))
            meaning = _declared_exit_meaning(core.root_cls, error_code)
            if meaning is not None:
                parts.append("exit code: %d (%s)" % (error_code, meaning))
            return _with_blob(
                _text_result("\n".join(p for p in parts if p), is_error=True),
                taken,
                name,
            )
        # The client only ever sees "Type: message"; the stack that says WHERE
        # the command broke exists nowhere else, so log it server-side too
        # (traceback under DUHO_TRACEBACK=1).
        _log_exception(
            _LOGGER,
            "tool %r raised: %s: %s",
            name,
            type(exc).__name__,
            exc,
        )
        return _text_result("%s: %s" % (type(exc).__name__, exc), is_error=True)
    finally:
        _exit_traceback_env(traceback_token)

    taken = capture.finish(output_var)
    if taken.error is not None:
        return taken.error
    stdout_text = taken.text

    def _result_for_stdout() -> dict:

        if isinstance(result, _Result) and (
            result.text is not None or result.value is not None
        ):
            words = result.text if result.text is not None else _render(result.value)
            parts = [stdout_text, words]
            return _text_result(
                "\n".join(part for part in parts if part), is_error=result.is_error
            )

        if isinstance(result, int):
            if result == 0:
                return _text_result(stdout_text, is_error=_is_error_flag(result))
            trailing = "exit code: %d" % result
            meaning = _declared_exit_meaning(core.root_cls, int(result))
            if meaning is not None:
                trailing += " (%s)" % meaning
            parts = [
                part for part in (stdout_text, err.getvalue().strip(), trailing) if part
            ]
            return _text_result("\n".join(parts), is_error=_is_error_flag(result, True))

        rendered = _render(result)
        if taken.wrote:
            rendered = stdout_text + "\n" + rendered
        return _text_result(rendered)

    return _with_blob(_result_for_stdout(), taken, name)

describe_tools(root_cls)

Describe every callable command in root_cls's tree as MCP tool specs.

root_cls is a Cmd/Cli class (the static _subcommands_ tree). The other form, a :class:_ServerCore, is the server core duho itself builds for an app() tree (the environment trigger and _mcp_command_); a caller does not construct one.

Each command reached by walking the built parser tree -- the root itself (when it can itself be dispatched), and every subcommand, recursively -- becomes one tool {name, description, inputSchema}: a leaf/root tool is named after the application (its root parser's prog, see :func:duho.args._app_name), a nested one parent.child. A NAMESPACE node (one whose own subcommand is mandatory -- see :func:_is_namespace_node) is skipped: it can never itself dispatch successfully, so listing it would only ever waste a client's turn on a guaranteed usage error; its own fields are still reachable, merged into every one of its descendants' schemas. This applies equally to a class command AND a module command (an app()-only concept -- every ModuleCommand node is itself always a leaf, never a namespace). A node whose class is (or subclasses) :class:McpCmd -- the self-serving command an app() may register under any name -- is likewise skipped (see :func:_is_mcp_command_node): serving one MCP session from inside a tool call another MCP session made makes no sense.

A node opted out via a per-command _mcp_ = False (see :attr:_Node.excluded, computed by :func:_walk_tree) is skipped too, together with its whole subtree -- the ROOT's own _mcp_ is exempt (never treated as this kind of exclusion).

Source code in src/duho/mcp/_tree.py

def describe_tools(root_cls: _ty.Union[type, _ServerCore]) -> list[dict]:
    """Describe every callable command in ``root_cls``'s tree as MCP tool specs.

    ``root_cls`` is a ``Cmd``/``Cli`` class (the static ``_subcommands_``
    tree). The other form, a :class:`_ServerCore`, is the server core duho
    itself builds for an ``app()`` tree (the environment trigger and
    ``_mcp_command_``); a caller does not construct one.

    Each command reached by walking the built parser tree -- the root itself
    (when it can itself be dispatched), and every subcommand, recursively --
    becomes one tool ``{name, description, inputSchema}``: a leaf/root tool is
    named after the application (its root parser's ``prog``, see
    :func:`duho.args._app_name`), a nested one ``parent.child``. A NAMESPACE
    node (one whose own subcommand is mandatory
    -- see :func:`_is_namespace_node`) is skipped: it can never itself
    dispatch successfully, so listing it would only ever waste a client's
    turn on a guaranteed usage error; its own fields are still reachable,
    merged into every one of its descendants' schemas. This applies equally
    to a class command AND a module command (an ``app()``-only concept --
    every ``ModuleCommand`` node is itself always a leaf, never a namespace).
    A node whose class is (or subclasses) :class:`McpCmd` -- the self-serving
    command an ``app()`` may register under any name -- is
    likewise skipped (see :func:`_is_mcp_command_node`): serving one MCP
    session from inside a tool call another MCP session made makes no sense.

    A node opted out via a per-command ``_mcp_ = False`` (see
    :attr:`_Node.excluded`, computed by :func:`_walk_tree`) is skipped too,
    together with its whole subtree -- the ROOT's own ``_mcp_`` is exempt
    (never treated as this kind of exclusion).
    """
    core = root_cls if isinstance(root_cls, _ServerCore) else _core_for_class(root_cls)
    return [
        _tool_spec(node)
        for node in core.nodes.values()
        if not _is_namespace_node(node.parser)
        and not _is_mcp_command_node(node)
        and not node.excluded
    ]

input_schema_for_command(cls)

Assemble a JSON-Schema object describing cls's own fields.

properties/required/additionalProperties: false from cls._getargs_() (each field's ArgumentBuilder, in declaration order) + duho._introspect.get_clsargs(cls) (the matching ClsArgDeclaration, for the raw annotation + docstring). Only cls's OWN fields -- not an inherited root's, not framework-injected argparse-only actions (--version/--help/--print-completion/--help-agents, none of which have an ArgumentBuilder behind them, so they're never reached by this per-builder walk at all).

This is the STANDALONE, single-class form (used directly, and by tests); the tool specs :func:describe_tools actually publishes for a NESTED command additionally merge in every ancestor's own fields and drop any field satisfiable from the server's environment/config -- see :func:_input_schema_for_node.

Source code in src/duho/mcp/_schema.py

def input_schema_for_command(cls: type[_Cmd]) -> dict:
    """Assemble a JSON-Schema ``object`` describing ``cls``'s own fields.

    ``properties``/``required``/``additionalProperties: false`` from
    ``cls._getargs_()`` (each field's ``ArgumentBuilder``, in declaration
    order) + ``duho._introspect.get_clsargs(cls)`` (the matching
    ``ClsArgDeclaration``, for the raw annotation + docstring). Only ``cls``'s
    OWN fields -- not an inherited root's, not framework-injected argparse-only
    actions (``--version``/``--help``/``--print-completion``/``--help-agents``,
    none of which have an ``ArgumentBuilder`` behind them, so they're never
    reached by this per-builder walk at all).

    This is the STANDALONE, single-class form (used directly, and by tests);
    the tool specs :func:`describe_tools` actually publishes for a NESTED
    command additionally merge in every ancestor's own fields and drop any
    field satisfiable from the server's environment/config -- see
    :func:`_input_schema_for_node`.
    """
    clsargs = _introspect.get_clsargs(cls)
    properties: dict = {}
    required: list[str] = []
    for builder in cls._getargs_():
        name = builder.name
        decl = clsargs.get(name)
        schema, is_required = json_schema_for_field(decl, builder)
        properties[name] = schema
        if is_required:
            required.append(name)
    return {
        "type": "object",
        "properties": properties,
        "required": required,
        "additionalProperties": False,
    }

json_schema_for_field(decl, builder)

Build (json_schema, required) for one field from its declaration + builder.

Consumes the same per-field data duho.agenthelp collects (a field's ClsArgDeclaration from duho._introspect.get_clsargs and its ArgumentBuilder from cls._getargs_()) without building a real argparse parser -- see the module docstring's "one real piece of work" deviation note for why (richer list[T]/dict[str, V] element-type fidelity than a stringified action type would give).

required is computed by :func:_is_required; when the field is not required, an explicit "default" is set on the schema (the field's real default when one exists, else None for the Optional[T]-with-no- default case) so an MCP client always sees what omitting the property resolves to. The description text is resolved by :func:_description_for.

Source code in src/duho/mcp/_schema.py

def json_schema_for_field(
    decl: _ty.Optional[_introspect.ClsArgDeclaration], builder: _ArgumentBuilder
) -> tuple[dict, bool]:
    """Build ``(json_schema, required)`` for one field from its declaration + builder.

    Consumes the same per-field data ``duho.agenthelp`` collects (a field's
    ``ClsArgDeclaration`` from ``duho._introspect.get_clsargs`` and its
    ``ArgumentBuilder`` from ``cls._getargs_()``) without building a real
    argparse parser -- see the module docstring's "one real piece of work"
    deviation note for why (richer ``list[T]``/``dict[str, V]`` element-type
    fidelity than a stringified ``action`` type would give).

    ``required`` is computed by :func:`_is_required`; when the field is not
    required, an explicit ``"default"`` is set on the schema (the field's real
    default when one exists, else ``None`` for the ``Optional[T]``-with-no-
    default case) so an MCP client always sees what omitting the property
    resolves to. The description text is resolved by :func:`_description_for`.
    """
    tp = decl.type if decl is not None and decl.type is not _NOT_DEFINED else None
    enum_by = getattr(builder, "enum_by", "name")
    schema = _schema_for_type(tp, enum_by) if tp is not None else {"type": "string"}

    if builder._kwargs().get("action") == "count":
        # Non-negative and capped: see `_MAX_COUNT_VALUE`.
        schema["maximum"] = _MAX_COUNT_VALUE
        schema["minimum"] = 0

    required = _is_required(builder)
    if not required:
        effective_default = builder._effective_default_()
        if effective_default is not _NOT_DEFINED:
            schema["default"] = _agenthelp._jsonable(effective_default, enum_by)
        else:
            schema.setdefault("default", None)

    help_text = _description_for(decl, builder)
    if help_text:
        schema["description"] = help_text

    return schema, required

main(argv=None)

python -m duho.mcp <app> entry point: resolve <app> and run :func:serve.

A duho CLI like any other: -h/--help and --version print to stdout and succeed, a missing <app>, an unknown option or an extra argument is a usage error (exit 2, nothing on stdout). <app> is a dotted qualname to a Cmd/Cli subclass (see :func:_resolve_app); one that does not resolve prints a one-line error to stderr and returns 1. Only then are the real stdio fds taken over for the protocol channel (see :func:_real_stdio_streams), before <app> is imported; the exit code is that of the stdio loop.

Source code in src/duho/mcp/_cli.py

def main(argv: _ty.Optional[_ty.Sequence[str]] = None) -> int:
    """``python -m duho.mcp <app>`` entry point: resolve ``<app>`` and run :func:`serve`.

    A duho CLI like any other: ``-h``/``--help`` and ``--version`` print to
    stdout and succeed, a missing ``<app>``, an unknown option or an extra
    argument is a usage error (exit ``2``, nothing on stdout). ``<app>`` is a
    dotted qualname to a ``Cmd``/``Cli`` subclass (see :func:`_resolve_app`);
    one that does not resolve prints a one-line error to stderr and returns
    ``1``. Only then are the real stdio fds taken over for the protocol
    channel (see :func:`_real_stdio_streams`), before ``<app>`` is imported;
    the exit code is that of the stdio loop.
    """
    return _main(_McpMain, argv)

serve(root_cls, *, stdin=None, stdout=None, output_var=None)

Run the stdio JSON-RPC loop for root_cls until stdin closes (EOF).

root_cls is a Cmd/Cli class (the static _subcommands_ tree). The :class:_ServerCore form is the server core duho builds for an app() tree (the environment trigger, _mcp_command_ and :func:serve_running_app); a caller does not construct one. Either is forwarded opaquely to :func:_handle_request, which in turn forwards it to :func:describe_tools/:func:call_tool (both already accept either shape -- see :func:describe_tools).

Reads newline-delimited JSON-RPC 2.0 request lines from stdin (real stdio, isolated per :func:_real_stdio_streams, when neither stdin nor stdout is given -- BINARY there, so one line's invalid UTF-8 bytes cannot kill the whole server, see below), dispatches each via :func:_handle_request, and writes any response line to stdout, flushed every time. A line that is not valid UTF-8 (only possible on the real-stdio path; an injected text stdin is decoded already) gets a -32700 parse-error response and the loop continues -- a client reconnecting or retrying is not required. A line that decodes but fails to parse as JSON gets the same -32700 (id: null in both cases -- the malformed line's own id, if any, is unrecoverable), as does a line whose {/[ nesting exceeds :data:_MAX_JSON_NESTING -- rejected BEFORE json.loads ever parses it, since parsing (or later re-encoding) a pathologically deep structure would otherwise raise an uncaught RecursionError and end the loop. A JSON ARRAY (a JSON-RPC 2.0 batch) is dispatched element by element; every non- notification element's response is collected into ONE reply array (never sent at all if the batch was all notifications, per spec), and an EMPTY batch array gets its own -32600. An unexpected exception from :func:_handle_request itself (which should never happen, given its own internal error handling, but must never end the server if it somehow does) becomes a -32603 response instead of propagating. Blank lines are skipped. Returns 0 when stdin reaches EOF (there is no separate MCP "shutdown" method to wait for). stdin/stdout are injectable so tests can drive the loop over in-memory TEXT streams instead of real pipes.

Before the first request it resolves where a tool's output that is not text goes (see :func:_resolve_output_target): descriptor 3 when open and inherited, or what the variable output_var names (default: the root's <NAME>_MCP variable plus _OUTPUT), a descriptor number or a file name opened for appending and closed when the loop ends. A bad variable writes one line to stderr and returns 2.

Source code in src/duho/mcp/_stdio.py

def serve(
    root_cls: _ty.Union[type, _ServerCore],
    *,
    stdin: _ty.Optional[_ty.TextIO] = None,
    stdout: _ty.Optional[_ty.TextIO] = None,
    output_var: _ty.Optional[str] = None,
) -> int:
    """Run the stdio JSON-RPC loop for ``root_cls`` until stdin closes (EOF).

    ``root_cls`` is a ``Cmd``/``Cli`` class (the static ``_subcommands_``
    tree). The :class:`_ServerCore` form is the server core duho builds for
    an ``app()`` tree (the environment trigger, ``_mcp_command_`` and
    :func:`serve_running_app`); a caller does not construct one. Either is
    forwarded opaquely to
    :func:`_handle_request`, which in turn forwards it to
    :func:`describe_tools`/:func:`call_tool` (both already accept either
    shape -- see :func:`describe_tools`).

    Reads newline-delimited JSON-RPC 2.0 request lines from ``stdin`` (real
    stdio, isolated per :func:`_real_stdio_streams`, when neither ``stdin``
    nor ``stdout`` is given -- BINARY there, so one line's invalid UTF-8
    bytes cannot kill the whole server, see below), dispatches each via
    :func:`_handle_request`, and writes any response line to ``stdout``,
    flushed every time. A line that is not valid UTF-8 (only possible on the
    real-stdio path; an injected text ``stdin`` is decoded already) gets a
    ``-32700`` parse-error response and the loop continues -- a client
    reconnecting or retrying is not required. A line that decodes but fails
    to parse as JSON gets the same ``-32700`` (``id: null`` in both cases --
    the malformed line's own id, if any, is unrecoverable), as does a line
    whose ``{``/``[`` nesting exceeds :data:`_MAX_JSON_NESTING` -- rejected
    BEFORE ``json.loads`` ever parses it, since parsing (or later
    re-encoding) a pathologically deep structure would otherwise raise an
    uncaught ``RecursionError`` and end the loop. A JSON ARRAY
    (a JSON-RPC 2.0 *batch*) is dispatched element by element; every non-
    notification element's response is collected into ONE reply array
    (never sent at all if the batch was all notifications, per spec), and an
    EMPTY batch array gets its own ``-32600``. An unexpected exception from
    :func:`_handle_request` itself (which should never happen, given its own
    internal error handling, but must never end the server if it somehow
    does) becomes a ``-32603`` response instead of propagating. Blank lines
    are skipped. Returns ``0`` when ``stdin`` reaches EOF (there is no
    separate MCP "shutdown" method to wait for). ``stdin``/``stdout`` are
    injectable so tests can drive the loop over in-memory TEXT streams
    instead of real pipes.

    Before the first request it resolves where a tool's output that is not text
    goes (see :func:`_resolve_output_target`): descriptor 3 when open and
    inherited, or what the variable ``output_var`` names (default: the root's
    ``<NAME>_MCP`` variable plus ``_OUTPUT``), a descriptor number or a file
    name opened for appending and closed when the loop ends. A bad variable
    writes one line to stderr and returns ``2``.
    """
    core_root = root_cls.root_cls if isinstance(root_cls, _ServerCore) else root_cls
    if output_var is None:
        output_var = _mcp_env_var_name(core_root) + "_OUTPUT"
    try:
        target = _resolve_output_target(_os.environ, output_var)
    except ValueError as exc:
        _compat.write_human(str(exc) + "\n", _sys.stderr)
        return 2
    if target is not None:
        # Offsets on a target that has no position count from this server's start.
        _WRITTEN.pop(target.fd, None)
    session: dict = {
        "output_fd": None if target is None else target.fd,
        "output_path": None if target is None else target.path,
        "output_var": output_var,
    }
    try:
        return _serve_stream(root_cls, stdin, stdout, session)
    finally:
        if target is not None and target.path is not None:
            _os.close(target.fd)

serve_running_app(transport='stdio')

Serve the CLI currently being dispatched as an MCP server, over transport (currently only "stdio").

Reads the app context :func:duho.args.main/:func:duho.runtime.app record in a ContextVar (duho._compat._MCP_CONTEXT) around their own dispatch step -- the exact parser/root class/dispatch callable THAT invocation already built -- so serving here needs no rediscovery: this is the SAME tree a client would see calling any other tool on the same running process, just entered from inside a dispatched command (e.g. :class:McpCmd) instead of the <PREFIX>MCP/<NAME>_MCP env trigger.

Output printed while the app was built (module-command imports, register() hooks) was written before this was called, so on this path it reaches stdout ahead of the first reply; the environment trigger takes stdio over before building and has no such output.

Raises RuntimeError when called outside such a dispatch (a bare script that never went through duho.main/duho.app at all has no running app context to serve). Raises ValueError for an unsupported transport -- checked BEFORE consulting the context, so it is reported the same way regardless of whether one exists.

Source code in src/duho/mcp/_stdio.py

def serve_running_app(transport: str = "stdio") -> int:
    """Serve the CLI currently being dispatched as an MCP server, over
    ``transport`` (currently only ``"stdio"``).

    Reads the app context :func:`duho.args.main`/:func:`duho.runtime.app`
    record in a ``ContextVar`` (``duho._compat._MCP_CONTEXT``) around their
    own dispatch step -- the exact parser/root class/dispatch callable THAT
    invocation already built -- so serving here needs no rediscovery: this
    is the SAME tree a client would see calling any other tool on the same
    running process, just entered from inside a dispatched command (e.g.
    :class:`McpCmd`) instead of the ``<PREFIX>MCP``/``<NAME>_MCP`` env
    trigger.

    Output printed while the app was built (module-command imports,
    ``register()`` hooks) was written before this was called, so on this path
    it reaches stdout ahead of the first reply; the environment trigger takes
    stdio over before building and has no such output.

    Raises ``RuntimeError`` when called outside such a dispatch (a bare
    script that never went through ``duho.main``/``duho.app`` at all has no
    running app context to serve). Raises ``ValueError`` for an unsupported
    ``transport`` -- checked BEFORE consulting the context, so it is
    reported the same way regardless of whether one exists.
    """
    if transport != "stdio":
        raise ValueError(
            "unsupported MCP transport %r (supported: stdio)" % (transport,)
        )
    ctx = _compat._MCP_CONTEXT.get()
    if ctx is None:
        raise RuntimeError(
            "serve_running_app() was called outside a duho.main()/duho.app() "
            "dispatch -- there is no currently-running app context to serve"
        )
    kind = ctx[0]
    if kind == "class":
        core = _core_for_class(ctx[1])
        output_var = None
    else:
        _, parser, root_cls, dispatch = ctx
        nodes = _walk_tree(parser, root_cls, parser.prog)
        core = _ServerCore(parser, nodes, dispatch, root_cls)
        output_var = _mcp_env_var_name(root_cls, name=parser.prog) + "_OUTPUT"
    return serve(core, output_var=output_var)