MCP

client → server  server/discover
server → client  {name: "orbit-releases", capabilities: {tools: {...}}}

client → server  tools/list
server → client  {tools: [{name: "get_release",
                           description: "Get the release date and notes ...",
                           input_schema: {type: "object", properties: {version: {type: "string"}},
                                          required: ["version"]}}, ...]}

client → server  tools/call {"name": "get_release", "arguments": {"version": "3.2.1"}}
server → client  {content: [{type: "text", text: "{\"version\": \"3.2.1\", ...}"}],
                  structured_content: {version: "3.2.1", date: "2026-10-20", notes: "Current"},
                  is_error: false}

# script_id: day_039_mcp_clients/remote_mcp_minimal
# OpenAI's servers call a public MCP server for you. Install: uv add "openai>=3.46,<4"
# Env: OPENAI_API_KEY, ATLAS_MODEL (default gpt-6-luna)   Services: none
import os

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()

DOCS_MCP = {
    "type": "mcp",
    "server_label": "openai_docs",  # your name for it; appears on output items
    "server_description": "Search and read the public OpenAI documentation.",
    "server_url": "https://developers.openai.com/mcp",
    "allowed_tools": ["search_openai_docs"],  # expose only what you need
    "require_approval": "never",  # read-only public docs; see the next section
}
resp = client.responses.create(model=MODEL, tools=[DOCS_MCP],
                               input="Find the Responses API streaming guide. Give the URL.")
print([item.type for item in resp.output])
print(resp.output_text)

['mcp_list_tools', 'mcp_call', 'message']
The streaming guide is at https://developers.openai.com/api/docs/guides/streaming-responses ...

# script_id: day_039_mcp_clients/remote_mcp_approval
# Default approvals: show each MCP call to a human before OpenAI makes it.
# Install: uv add "openai>=3.46,<4"   Env: OPENAI_API_KEY, ATLAS_MODEL   Services: none
import os

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()
TOOLS = [{"type": "mcp", "server_label": "openai_docs",
          "server_url": "https://developers.openai.com/mcp",
          "allowed_tools": ["search_openai_docs"]}]  # no require_approval -> approval required

def ask_with_approvals(question: str, max_rounds: int = 3) -> str:
    resp = client.responses.create(model=MODEL, tools=TOOLS, input=question)
    for _ in range(max_rounds):
        requests = [i for i in resp.output if i.type == "mcp_approval_request"]
        if not requests:
            return resp.output_text
        answers = []
        for req in requests:  # arguments is a JSON string the model wrote
            ok = input(f"Allow {req.server_label}.{req.name}({req.arguments})? [y/N] ")
            answers.append({"type": "mcp_approval_response", "approval_request_id": req.id,
                            "approve": ok.strip().lower() == "y"})
        # same tools again; previous_response_id carries the earlier items
        resp = client.responses.create(model=MODEL, tools=TOOLS,
                                       previous_response_id=resp.id, input=answers)
    return "(stopped: too many approval rounds)"

print(ask_with_approvals("Find the Responses API streaming guide."))

Allow openai_docs.search_openai_docs({"query": "Responses API streaming"})? [y/N] y
The streaming guide is at https://developers.openai.com/api/docs/guides/streaming-responses ...

# script_id: day_039_mcp_clients/orbit_release_server
# Provided black-box MCP server: Orbit releases + error codes (read-only). You build one on Day 40.
# Install: uv add "mcp[cli]"   Env: none   Services: none (a host launches this over stdio)
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations
from pydantic import BaseModel

RELEASES = {  # version -> (date, notes), from DATASET_SPEC §1
    "2.9.0": ("2025-11-20", "Last 2.x; EOL 2026-12-31"),
    "3.0.0": ("2026-03-02", "Introduced Flows and orbit.toml (replacing orbit.ini + ORBIT_* env vars)"),
    "3.1.0": ("2026-06-01", ""), "3.2.0": ("2026-09-15", ""),
    "3.2.1": ("2026-10-20", "Current"),
}
ERROR_CODES = {"ORB-1101": "LeaseExpired", "ORB-1102": "WorkerHeartbeatLost",
               "ORB-1201": "SerializationError", "ORB-1202": "PayloadTooLarge",
               "ORB-1301": "RetryExhausted", "ORB-1901": "BrokerRateLimited"}

class Release(BaseModel):
    version: str
    date: str
    notes: str

mcp = MCPServer("orbit-releases", version="1.0.0")

@mcp.tool(annotations=ToolAnnotations(read_only_hint=True))
def get_release(version: str) -> Release:
    """Get the release date and notes for one Orbit version, e.g. '3.2.1'."""
    if version not in RELEASES:
        raise ToolError(f"Unknown version {version!r}. Known: {', '.join(RELEASES)}")
    date, notes = RELEASES[version]
    return Release(version=version, date=date, notes=notes)

@mcp.tool(annotations=ToolAnnotations(read_only_hint=True))
def list_error_codes(prefix: str = "ORB-") -> dict[str, str]:
    """List Orbit error codes whose code starts with prefix, e.g. 'ORB-11' for lease errors."""
    return {code: name for code, name in ERROR_CODES.items() if code.startswith(prefix)}

if __name__ == "__main__":
    mcp.run()  # stdio: stdout carries the protocol, so never print() here

# script_id: day_039_mcp_clients/local_client
# Launch the Orbit release server over stdio, list its tools, call one.
# Install: uv add "mcp[cli]"   Env: none   Services: none (no API key needed)
import asyncio
import sys
from pathlib import Path

from mcp import Client, StdioServerParameters

SERVER = Path(__file__).with_name("orbit_release_server.py")

async def main() -> None:
    # sys.executable = the same Python (and venv) that runs this client
    params = StdioServerParameters(command=sys.executable, args=[str(SERVER)])
    async with Client(params) as client:
        print("protocol:", client.protocol_version)
        listed = await client.list_tools()
        print([t.name for t in listed.tools])
        print(listed.tools[0].input_schema)
        result = await client.call_tool("get_release", {"version": "3.2.1"})
        print(result.is_error, result.structured_content)
        bad = await client.call_tool("get_release", {"version": "9.9"})
        print(bad.is_error, bad.content[0].text)

asyncio.run(main())

protocol: 2026-07-28
['get_release', 'list_error_codes']
{'type': 'object', 'properties': {'version': {'title': 'Version', 'type': 'string'}}, 'required': ['version'], 'title': 'get_releaseArguments'}
False {'version': '3.2.1', 'date': '2026-10-20', 'notes': 'Current'}
True Error executing tool get_release: Unknown version '9.9'. Known: 2.9.0, 3.0.0, 3.1.0, 3.2.0, 3.2.1

# script_id: day_039_mcp_clients/mcp_bridge
# Bridge MCP servers into the Day 35 tool loop. Install: uv add "mcp[cli]" "openai>=3.46,<4"
# Env: OPENAI_API_KEY, ATLAS_MODEL (default gpt-6-luna), ORBIT_DATA (default datasets/orbit)
import asyncio
import json
import os
import sys
from contextlib import AsyncExitStack
from dataclasses import dataclass, field
from pathlib import Path

from mcp import Client, StdioServerParameters
from openai import AsyncOpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
SEP = "__"  # server prefix separator: orbit__get_release

def envelope_err(kind: str, message: str, hint: str | None = None) -> dict:  # compact Day 37 envelope
    return {"ok": False, "error": {"kind": kind, "message": message, "hint": hint}}

def search_docs(query: str, k: int = 3) -> list[str]:  # compact Day 36 tool: keyword scorer
    words = query.lower().split()
    scores = {p.stem: sum(p.read_text().lower().count(w) for w in words)
              for p in (DATA_DIR / "docs").glob("*.md")}
    return sorted((d for d in scores if scores[d]), key=scores.get, reverse=True)[:k]

LOCAL_TOOLS = [{"type": "function", "name": "search_docs", "strict": True,
                "description": "Find Orbit documentation pages by keyword. Returns page ids.",
                "parameters": {"type": "object", "additionalProperties": False,
                               "properties": {"query": {"type": "string"}, "k": {"type": "integer"}},
                               "required": ["query", "k"]}}]

async def mcp_tools_as_functions(client: Client, prefix: str) -> list[dict]:
    """Turn one server's MCP tools into Responses function tools named <prefix>__<tool>."""
    listed = await client.list_tools()
    return [{"type": "function", "name": f"{prefix}{SEP}{t.name}",
             "description": t.description or "", "parameters": t.input_schema,
             "strict": False}  # MCP schemas carry defaults; the server validates
            for t in listed.tools]

# script_id: day_039_mcp_clients/mcp_bridge
async def call_mcp_tool(client: Client, name: str, args: dict) -> str:
    """Call one MCP tool; return JSON for the model. Never raises."""
    try:
        result = await client.call_tool(name, args, read_timeout_seconds=10.0)
    except Exception as exc:  # transport failure or timeout, not a tool error
        return json.dumps(envelope_err("transient", f"{type(exc).__name__}: {exc}"))
    text = " ".join(getattr(c, "text", "") for c in result.content)
    if result.is_error:
        return json.dumps(envelope_err("tool_error", text, hint="Fix the arguments or try another tool."))
    if result.structured_content is not None:
        return json.dumps({"ok": True, "data": result.structured_content})
    return json.dumps({"ok": True, "data": text})

async def dispatch(name: str, arguments: str, clients: dict[str, Client]) -> str:
    """Route a function_call: <prefix>__<tool> goes to MCP, anything else is local."""
    args = json.loads(arguments or "{}")
    prefix, sep, tool = name.partition(SEP)
    if sep and prefix in clients:
        return await call_mcp_tool(clients[prefix], tool, args)
    if name == "search_docs":
        return json.dumps({"ok": True, "data": search_docs(**args)})
    return json.dumps(envelope_err("unknown_tool", f"No tool named {name!r}"))

# script_id: day_039_mcp_clients/mcp_bridge
@dataclass
class LoopResult:  # same shape as Day 35
    text: str
    rounds: int
    stop_reason: str
    usage: dict = field(default_factory=lambda: {"input_tokens": 0, "output_tokens": 0})
    trace: list[str] = field(default_factory=list)

async def run_mcp_tool_loop(question: str, clients: dict[str, Client], *,
                            extra_tools: list[dict] | None = None, max_rounds: int = 5) -> LoopResult:
    tools = list(extra_tools or [])
    for prefix, client in clients.items():
        tools += await mcp_tools_as_functions(client, prefix)
    llm = AsyncOpenAI()
    items: list = [{"role": "user", "content": question}]
    result = LoopResult(text="", rounds=0, stop_reason="max_rounds")
    for rnd in range(1, max_rounds + 1):
        result.rounds = rnd
        resp = await llm.responses.create(model=MODEL, input=items, tools=tools)
        result.usage["input_tokens"] += resp.usage.input_tokens  # feed Day 12's CostTracker here
        result.usage["output_tokens"] += resp.usage.output_tokens
        calls = [i for i in resp.output if i.type == "function_call"]
        if not calls:
            result.text, result.stop_reason = resp.output_text, "answered"
            return result
        items += resp.output
        for call in calls:  # answer every call before the next request
            output = await dispatch(call.name, call.arguments, clients)
            result.trace.append(f"round {result.rounds}: {call.name}({call.arguments}) -> {output[:60]}")
            items.append({"type": "function_call_output", "call_id": call.call_id, "output": output})
    return result

# script_id: day_039_mcp_clients/mcp_bridge
# one entry per server; orbit_release_server.py must sit beside this file
SERVERS = {"orbit": StdioServerParameters(
    command=sys.executable, args=[str(Path(__file__).with_name("orbit_release_server.py"))])}

async def main() -> None:
    if not (DATA_DIR / "docs").is_dir():  # fail fast: a wrong cwd would make search_docs return []
        sys.exit(f"Orbit docs not found at {DATA_DIR / 'docs'}; set ORBIT_DATA to the dataset folder")
    async with AsyncExitStack() as stack:
        clients = {name: await stack.enter_async_context(Client(params))
                   for name, params in SERVERS.items()}
        result = await run_mcp_tool_loop(
            "When was Orbit 3.0.0 released, what did it change, and which doc page covers migrating?",
            clients, extra_tools=LOCAL_TOOLS)
        print("\n".join(result.trace))
        print(f"[{result.stop_reason} in {result.rounds} rounds, {result.usage}]")
        print(result.text)

if __name__ == "__main__":
    asyncio.run(main())

round 1: orbit__get_release({"version":"3.0.0"}) -> {"ok": true, "data": {"version": "3.0.0", "date": "2026-03-0
round 1: search_docs({"query":"migrating 2 to 3","k":3}) -> {"ok": true, "data": ["changelog", "migrating-2-to-
[answered in 2 rounds, {'input_tokens': 1312, 'output_tokens': 164}]
Orbit 3.0.0 was released on 2026-03-02. It introduced Flows and replaced orbit.ini ... see migrating-2-to-3.

# script_id: day_039_mcp_clients/registry_search
# Search the official MCP registry (preview, read-only, no auth). Install: none (stdlib only)
# Env: none   Services: network access to registry.modelcontextprotocol.io
import json
import sys
import urllib.error
import urllib.parse
import urllib.request

REGISTRY = "https://registry.modelcontextprotocol.io/v0.1/servers"

def search_registry(term: str, limit: int = 5) -> list[dict]:
    """term is a name substring (3+ characters); version=latest skips old versions."""
    query = urllib.parse.urlencode({"search": term, "version": "latest", "limit": limit})
    with urllib.request.urlopen(f"{REGISTRY}?{query}", timeout=15) as resp:
        return json.load(resp).get("servers", [])

if __name__ == "__main__":
    term = sys.argv[1] if len(sys.argv) > 1 else "git"
    try:
        entries = search_registry(term)
    except urllib.error.URLError as exc:
        sys.exit(f"registry unreachable: {exc.reason}")
    for entry in entries:
        server = entry.get("server", entry)  # preview API: if fields move, print(entry) and adapt
        packages = [f"{p.get('registryType')}:{p.get('identifier')}" for p in server.get("packages", [])]
        print(server.get("name"), server.get("version"), packages)
        print("   ", (server.get("description") or "")[:90])

ai.smithery/Hint-Services-obsidian-github-mcp 0.4.0 []
    Connect AI assistants to your GitHub-hosted Obsidian vault to seamlessly access, search, a
ai.smithery/smithery-ai-github 1.0.0 []
    Access the GitHub API, enabling file operations, repository management, search functionali
...