MCP servers

import os
import sys
from pathlib import Path

from shipit_agent import Agent
from shipit_agent.llms import build_llm_from_env
from shipit_agent.llms.factory import build_llm_from_settings

os.environ["SHIPIT_ANTHROPIC_MODEL"] = "claude-haiku-5-5"
haiku = build_llm_from_env("anthropic")
gemma = build_llm_from_settings({"model": "google.gemma-4-31b"}, provider="bedrock")

SERVER = '\n"""A tiny MCP server over stdio: team notes, with tools, resources and a prompt."""\nimport json\nimport sys\nfrom pathlib import Path\n\nSTORE = Path("notes.json")\nDEFAULT_NOTES = {\n    "deploy": "Deploys go out Tuesdays and Thursdays at 14:00 UTC. Freeze the day before a holiday.",\n    "oncall": "On-call rotates weekly on Monday. Page #oncall first, then the secondary.",\n    "release": "Release notes live in CHANGELOG.md; tag as vX.Y.Z after the notes merge.",\n}\nTOOLS = [\n    {"name": "search_notes", "description": "Search the team notes by keyword and return matching notes.",\n     "inputSchema": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},\n     "annotations": {"readOnlyHint": True}},\n    {"name": "add_note", "description": "Add or replace a team note under a topic.",\n     "inputSchema": {"type": "object", "properties": {"topic": {"type": "string"}, "text": {"type": "string"}},\n                     "required": ["topic", "text"]}},\n]\n\n\ndef load():\n    return json.loads(STORE.read_text()) if STORE.exists() else dict(DEFAULT_NOTES)\n\n\ndef handle(method, params):\n    NOTES = load()\n    if method == "initialize":\n        return {"protocolVersion": "2025-06-18", "serverInfo": {"name": "notes", "version": "1.0"},\n                "capabilities": {"tools": {}, "resources": {}, "prompts": {}},\n                "instructions": "Team notes. Search before answering process questions."}\n    if method == "tools/list":\n        return {"tools": TOOLS}\n    if method == "tools/call":\n        name, args = params["name"], params.get("arguments", {})\n        if name == "search_notes":\n            q = args["query"].lower()\n            hits = {k: v for k, v in NOTES.items() if q in k or q in v.lower()}\n            text = json.dumps(hits) if hits else f"no note mentions {args[\'query\']!r}"\n        else:\n            NOTES[args["topic"]] = args["text"]\n            STORE.write_text(json.dumps(NOTES))\n            text = f"saved note {args[\'topic\']!r}"\n        return {"content": [{"type": "text", "text": text}], "isError": False}\n    if method == "resources/list":\n        return {"resources": [{"uri": f"notes://{k}", "name": k, "mimeType": "text/plain"} for k in NOTES]}\n    if method == "resources/read":\n        key = params["uri"].removeprefix("notes://")\n        return {"contents": [{"uri": params["uri"], "mimeType": "text/plain", "text": NOTES.get(key, "")}]}\n    if method == "prompts/list":\n        return {"prompts": [{"name": "handover", "description": "Write an on-call handover.",\n                             "arguments": [{"name": "who", "required": True}]}]}\n    if method == "prompts/get":\n        who = params.get("arguments", {}).get("who", "the next person")\n        return {"messages": [{"role": "user", "content": {"type": "text",\n                "text": f"Write a short on-call handover for {who}, using the oncall note."}}]}\n    if method == "ping":\n        return {}\n    raise ValueError(f"unsupported method {method}")\n\n\nfor line in sys.stdin:\n    request = json.loads(line)\n    if "id" not in request:\n        continue\n    try:\n        reply = {"result": handle(request["method"], request.get("params", {}))}\n    except Exception as error:\n        reply = {"error": {"code": -32601, "message": str(error)}}\n    print(json.dumps({"jsonrpc": "2.0", "id": request["id"], **reply}), flush=True)\n'
Path("notes_server.py").write_text(SERVER)
print(len(SERVER.splitlines()), "lines written to notes_server.py")

from shipit_agent import MCPStdioTransport, RemoteMCPServer

notes = RemoteMCPServer(name="notes", transport=MCPStdioTransport(command=[sys.executable, "notes_server.py"]))
for tool in notes.discover_tools():
    print(f"{tool.name:13} {tool.description}")
print("server:", notes.server_info, "| says:", notes.instructions)

from shipit_agent.tools.base import ToolContext

search = next(t for t in notes.discover_tools() if t.name == "search_notes")
print(search.run(ToolContext(prompt="demo"), query="deploy").text)

agent = Agent(llm=haiku, prompt="Answer from the team notes. Be brief.", mcps=[notes])
result = agent.run("When do deploys go out?")
print(result.output)
print("calls:", [(r.name, r.output[:60]) for r in result.tool_results])

from datetime import date

from shipit_agent.tools import FunctionTool

def today() -> str:
    """Today's date and weekday."""
    return date(2026, 11, 9).strftime("%A %Y-%m-%d")

agent = Agent(llm=haiku, prompt="Answer from the team notes. Be brief.", mcps=[notes],
              tools=[FunctionTool.from_callable(today, read_only=True)])
result = agent.run("Is there a deploy today? Check the date and the notes.")
print(result.output)
print("calls:", [r.name for r in result.tool_results])

writer = Agent(llm=haiku, prompt="Keep the team notes current. Be brief.", mcps=[notes],
               permission_mode="bypassPermissions")
print(writer.run("Add a note under 'freeze': no deploys from Dec 20 to Jan 3.").output)
print(search.run(ToolContext(prompt="demo"), query="freeze").text)

read_only = RemoteMCPServer(name="notes", transport=MCPStdioTransport(command=[sys.executable, "notes_server.py"]),
                            allowed_tools={"search_notes"})
print([t.name for t in read_only.discover_tools()])

named = RemoteMCPServer(name="notes", transport=MCPStdioTransport(command=[sys.executable, "notes_server.py"]),
                        include_server_in_tool_names=True)
print([t.name for t in named.discover_tools()])
print(Agent(llm=haiku, prompt="Be brief.", mcps=[named]).run("What does the oncall note say?").output)

for r in notes.list_resources():
    print(r.uri, "|", r.name)
print(notes.read_resource("notes://release"))

print([p.name for p in notes.list_prompts()])
handover = notes.get_prompt("handover", {"who": "Priya"})
print(handover)
print("\n" + Agent(llm=haiku, mcps=[notes]).run(handover).output)

for e in Agent(llm=haiku, prompt="Be brief.", mcps=[notes]).stream("Who do I page first when on call?"):
    if e.type == "tool_called":
        print("call", e.payload["tool"], e.payload["arguments"])
    elif e.type == "run_completed":
        print(e.payload["output"])

result = Agent(llm=gemma, prompt="Answer from the team notes. Be brief.", mcps=[notes]).run(
    "Where do release notes go, and how do we tag a release?")
print(result.output)
print("calls:", [r.name for r in result.tool_results])

from shipit_agent import list_mcp_catalog

catalog = list_mcp_catalog()
print(len(catalog), "servers")
for c in catalog[:12]:
    how = "hosted" if c.url else "stdio "
    print(f"{c.name:16} {how} {c.auth:5} {c.description}")

for server in (notes, read_only, named):
    server.close()
Path("notes_server.py").unlink()
Path("notes.json").unlink(missing_ok=True)