server.py

"""Multi-round tool result (2026 era): a tool returns input_required and resumes from echoed state."""

from mcp_types import ElicitRequest, ElicitRequestedSchema, ElicitRequestFormParams, ElicitResult, InputRequiredResult

from mcp.server.mcpserver import Context, MCPServer
from stories._hosting import run_server_from_args

CONFIRM_SCHEMA: ElicitRequestedSchema = {
    "type": "object",
    "properties": {"confirm": {"type": "boolean", "description": "Proceed with the deployment?"}},
    "required": ["confirm"],
}


def build_server() -> MCPServer:
    # requestState is sealed by default under a process-local key, which suits this
    # single-process server; fleets share keys=[...] so any instance can verify.
    mcp = MCPServer("mrtr-example")

    @mcp.tool(description="Deploy to an environment, asking the user to confirm first.")
    async def deploy(env: str, ctx: Context) -> str | InputRequiredResult:
        responses = ctx.input_responses
        if responses is None or "confirm" not in responses:
            ask = ElicitRequest(
                params=ElicitRequestFormParams(message=f"Deploy to {env}?", requested_schema=CONFIRM_SCHEMA)
            )
            # The boundary seals this plaintext request_state on the way out and unseals the echo on retry.
            return InputRequiredResult(input_requests={"confirm": ask}, request_state="awaiting-confirm")
        assert ctx.request_state == "awaiting-confirm", ctx.request_state
        answer = responses["confirm"]
        if isinstance(answer, ElicitResult) and answer.action == "accept" and (answer.content or {}).get("confirm"):
            return f"deployed to {env}"
        return f"deployment to {env} cancelled"

    return mcp


if __name__ == "__main__":
    run_server_from_args(build_server)