MCP: advanced topics

Normally the client calls the server. Sampling flips it: the server asks the client's model to do some thinking for it. This tool reads a file and asks the client's Claude to summarize it — so the server never needs its own API key or model choice.

1. from mcp.server.fastmcp import FastMCP, Context imports the real MCP SDK. FastMCP("summarizer") creates a named server, and Context is the handle that lets a tool talk back to the client.
2. @mcp.tool() registers the function below as a tool the client can call. async def means it can pause while waiting on the network without blocking everything else.
3. text = open(path).read()[:4000] reads the file and keeps only the first 4000 characters — a simple guard so a huge file doesn't blow past the model's limits.
4. await ctx.session.create_message(...) is the sampling call: the server asks the client's model for a completion, passing a normal messages list and max_tokens=200. The client runs its model and hands the answer back.
5. return result.content.text sends the model's summary back as the tool's result.

What the output means: Calling this tool returns a short summary of the file, produced by the client's model — even though the server itself has no API key.

Try this: Notice the server never picks a model or pays for tokens — the client does, and it can approve or deny each sampling request. That single approval path is exactly why sampling is considered safe.

A notification is a one-way heads-up from server to client: "something changed, refresh yourself." Without it the client would have to keep asking "anything new yet?" (polling). Here, adding a document tells the client its list of resources is out of date.

1. def save_document(name, body): ... is a placeholder for your real storage — the ... just means "fill this in." The lesson focuses on the notification, not the saving.
2. @mcp.tool() makes add_document a callable tool. It takes a name and body plus the ctx handle to reach the client.
3. save_document(name, body) stores the new document (in real code).
4. await ctx.session.send_resource_list_changed() is the notification: it tells the client "the set of resources changed, so what you cached is stale — reload it." The client refreshes on its own; the server never waits for a reply.
5. return f"added {name}" gives the caller a plain confirmation.

What the output means: The document is saved and the client is nudged to refresh its resource list automatically — no polling loop needed.

Try this: Compare this to refreshing a webpage: instead of you hitting reload over and over, the server says "new content is ready" and the page updates itself.

A server has to speak over a transport — the wire between client and server. This tiny launcher lets the same server run either as a local program (stdio) or as a network service (streamable-http), chosen by a command-line flag.

1. mcp = FastMCP("docs") builds the server object; the comment notes the tools were registered earlier in the file.
2. if __name__ == "__main__": means "only run this part when this file is executed directly" — the standard Python way to add a startup section.
3. import sys gives access to sys.argv, the list of words typed after the program name on the command line.
4. "--http" in sys.argv checks whether you launched with the --http flag. If yes, transport="streamable-http" (networked, multi-client); if no, "stdio" (a local subprocess the client starts). mcp.run(...) then starts the server that way.

What the output means: Run it plainly and it starts on stdio for a local client; run it with --http and the same code serves over the network.

Try this: Run the file two ways — with and without --http — and notice you changed the transport without touching any tool code. That separation is the whole point.

Roots are the folders the client says a server is allowed to touch. This code is the security check a filesystem server must run before opening any file: it confirms the requested path really lives inside an allowed root and isn't a sneaky attempt to escape it. This block actually runs and prints its results.

1. os.path.realpath(path) resolves the true, absolute location — it follows shortcuts and collapses tricks like ../ (go up a folder). This is the crucial step: you must compare the real destination, not the text the caller typed.
2. real.startswith(os.path.realpath(r) + os.sep) checks the resolved path begins with an allowed root folder (the + os.sep makes sure /home/user/project2 can't sneak past a root of /home/user/project). any(...) returns true if the path is inside any allowed root.
3. The for loop tests three paths: a legitimate file inside the root, a ../secrets.txt path-escape attempt, and an unrelated system file /etc/passwd.
4. print(f"{within_roots(p, roots)!s:5} {p}") prints the True/False verdict (padded to width 5 with !s:5) next to each path.

What the output means: Only the file genuinely inside /home/user/project prints True; the ../secrets.txt escape and /etc/passwd both print False — blocked.

Try this: Add "/home/user/project/deep/nested/file.txt" to the list — it passes, because it's genuinely inside the root. Then try "/home/user" (the parent) — it fails. That's the boundary doing its job.

A production tool should never crash the client — it should catch its own problems and return a tidy result either way. This wrapper runs any function, and instead of letting an error explode, it hands back a small dictionary saying whether it succeeded. This is the same resiliency habit you'd give any real service.

1. def safe_tool(fn, *args, timeout_hit=False): takes the function to run (fn), whatever arguments it needs (*args), and a test flag to simulate a timeout.
2. try: attempts the work. If timeout_hit is set it raises TimeoutError to mimic a slow operation; otherwise it runs fn(*args) and returns {"ok": True, "result": ...}.
3. except TimeoutError catches the slow-operation case and returns {"ok": False, "error": "timeout: ..."} — a clean message, not a hang.
4. except Exception as e is the safety net for any other failure. type(e).__name__ names the error kind (like AttributeError) so the caller learns what went wrong.
5. The three print lines exercise all paths: a success, a real bug (calling .upper() on the number 42), and a simulated timeout.

What the output means: Three dictionaries print: the first has ok: True with 'HELLO'; the other two have ok: False with a readable error string — never a crash.

Try this: Every result has the same shape — an ok flag plus either result or error. That predictable shape is what lets the client keep working no matter how the tool fails.