A tool is only one of the three things an MCP server can offer. A code assistant that answers questions about a library also needs to attach the right docs page to the conversation without a model turn, and to let a person start a review of a page from a menu. MCP has a separate primitive for each job: tools, resources and prompts. You will build one docs server that offers all three with the official MCP Python SDK, mcp 2.3.0, and test the error code each one returns when a request is wrong.
When you need this. Your team’s internal library has a coding assistant. Searching the docs is a call the model decides to make, but attaching the page the user has open should not cost a model turn, and a reviewer wants a ready-made “review this page” action in the host’s menu. Those three jobs map to the three primitives, and one server can offer all of them.
1. Set up the project
Download the complete lab ZIP, which includes the Python files, fixtures, tests and README. Extract it before running the commands below.
unzip build-an-mcp-server-with-tools-resources-and-prompts.zip
cd mcp-server-primitives
python3.12 -m venv .venv
.venv/bin/python -m pip install -r requirements-lock.txtThe server reads three pages of sample docs for taskq, a job-queue library that exists only in this lesson:
{
"project": "taskq, a sample job-queue library used only in this lesson",
"pages": {
"quickstart": {
"title": "Quickstart",
"body": "# Quickstart\n\nCreate a queue with `Queue(\"emails\")` and add a job with `queue.put(job)`. Start a worker with `taskq worker emails`. A worker takes one job at a time and acknowledges it when the handler returns.\n"
},
"configuration": {
"title": "Configuration",
"body": "# Configuration\n\nSet `TASKQ_BROKER_URL` to the broker address. `visibility_timeout` is the number of seconds a job stays hidden after a worker takes it; the default is 30. Set it longer than your slowest handler, or a second worker takes the same job.\n"
},
"retries": {
"title": "Retries",
"body": "# Retries\n\nA handler that raises is retried up to `max_retries` times, 3 by default, with exponential backoff. After the last attempt the job moves to the dead-letter queue. Make handlers idempotent: a retried job can run twice.\n"
}
}
}2. Write the server
Now we write the whole server, in one file. Each decorator registers one primitive, and the SDK generates the listings, schemas and capabilities from the Python signatures.
"""A docs server: one search tool, the docs pages as resources and a review prompt.
Run over stdio with `python docs_server.py`. The pages come from docs.json.
Built with the official MCP Python SDK (mcp 2.3.0).
"""
import json
from pathlib import Path
from typing import TypedDict
from mcp import MCPError
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ResourceNotFoundError, ToolError
from mcp.server.mcpserver.prompts.base import UserMessage
from mcp.types import INVALID_PARAMS, EmbeddedResource, TextResourceContents, ToolAnnotations
PAGES = json.loads((Path(__file__).with_name("docs.json")).read_text(encoding="utf-8"))["pages"]
class DocsServer(MCPServer):
"""Report an unknown prompt or a missing prompt argument as -32602, as the prompts specification asks."""
async def get_prompt(self, name, arguments=None, context=None):
prompt = next((p for p in await self.list_prompts() if p.name == name), None)
if prompt is None:
raise MCPError(INVALID_PARAMS, f"Unknown prompt: {name}")
missing = [a.name for a in prompt.arguments or [] if a.required and a.name not in (arguments or {})]
if missing:
raise MCPError(INVALID_PARAMS, f"Missing required arguments: {', '.join(missing)}")
return await super().get_prompt(name, arguments, context)
server = DocsServer("docs", version="1.0.0", instructions="Search and read the taskq documentation.")
class SearchHit(TypedDict):
slug: str
title: str
uri: str
@server.tool(title="Search the docs", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False))
def search_docs(query: str) -> list[SearchHit]:
"""Return the docs pages whose text contains every word of the query."""
words = query.lower().split()
if not words:
raise ToolError("The query is empty. Pass one or more words, such as 'worker timeout'.")
return [{"slug": slug, "title": page["title"], "uri": f"docs://pages/{slug}"}
for slug, page in PAGES.items() if all(word in page["body"].lower() for word in words)]
@server.resource("docs://index", title="Docs index", mime_type="application/json")
def docs_index() -> str:
"""Every page's slug and title."""
return json.dumps([{"slug": slug, "title": page["title"]} for slug, page in PAGES.items()])
@server.resource("docs://pages/{slug}", title="Docs page", mime_type="text/markdown")
def docs_page(slug: str) -> str:
"""The Markdown text of one docs page."""
if slug not in PAGES:
raise ResourceNotFoundError(f"No docs page named {slug}.")
return PAGES[slug]["body"]
@server.prompt(title="Review a docs page")
def review_page(slug: str, focus: str = "accuracy") -> list[UserMessage]:
"""Ask the model to review one docs page; the page text is embedded in the prompt."""
if slug not in PAGES:
raise MCPError(INVALID_PARAMS, f"No docs page named {slug}.")
page = TextResourceContents(uri=f"docs://pages/{slug}", mime_type="text/markdown", text=PAGES[slug]["body"])
return [UserMessage(EmbeddedResource(type="resource", resource=page)),
UserMessage(f"Review this docs page for {focus}. List each problem with the sentence it is in.")]
if __name__ == "__main__":
server.run()The MCP specification describes the three primitives by who controls them, and that is the rule for choosing one. MCP server features
| Primitive | Controlled by | In docs_server.py |
|---|---|---|
| Tool | The model, which proposes calls | search_docs: find pages that contain the query words |
| Resource | The application, which decides what context to attach | docs://index and each page at docs://pages/{slug} |
| Prompt | The user, who picks it, for example as a slash command | review_page: start a review of one page |
3. Return search results the client can validate
search_docs returns a list of the TypedDict SearchHit. The SDK turns that return type into the tool’s outputSchema and sends matching JSON in structuredContent, so a client can validate the result instead of parsing text. The annotations readOnlyHint: true and openWorldHint: false tell the client that the tool reads only its own data. An empty query raises ToolError, which the SDK sends as a result with isError: true that the model can read and correct. MCP tools
The recorded search for worker returned 2 pages: quickstart, configuration.
Note: annotations are hints from the server, not permissions. The schema says annotations from untrusted servers must not decide tool use, so keep your own allowlist. MCP schema 2026-07-28: ToolAnnotations
Search with an empty query. The server returns isError: true. Ask for a nonempty search query instead of parsing an empty result as a successful search.
4. Expose the pages as resources
A resource is data identified by a URI and read with resources/read. @server.resource("docs://index") registers one fixed resource, the list of pages. @server.resource("docs://pages/{slug}") has a placeholder, so the SDK lists it as a resource template, an RFC 6570 URI template, and passes the matched part of the URI to docs_page as slug. Each read returns contents with the URI, the MIME type and the text. MCP resources
Templates keep the listing small: resources/list returned 1 resource, and every page stays reachable through the template. A docs site with thousands of pages can list none of them and let the client fill in the template.
For a page that does not exist, the specification requires JSON-RPC error -32602 and forbids an empty contents array. docs_page raises ResourceNotFoundError, which the SDK sends as -32602. Its base class, ResourceError, would go out as -32603, an internal error, which tells the client the server broke.
Page that does not exist. docs_page raises ResourceNotFoundError, which reaches the client as -32602, not empty contents. Let the application ask for another page or report that it is missing.
URI that matches no resource or template. The SDK returns -32602 Unknown resource: docs://drafts/retries before calling a handler. Build URIs from the published template.
A template lets any client name any slug. That is fine for public docs. For private pages, check the caller’s permission inside docs_page and review_page, and return the same -32602 for a page the caller cannot see as for one that does not exist, so the error does not reveal which pages exist.
5. Add a prompt that embeds a page
A prompt is a message template with named arguments. prompts/get fills in the arguments and returns messages ready for the application to send to a model. review_page takes a required slug and an optional focus, which defaults to accuracy. Its first message is an embedded resource: the page itself, with its URI and MIME type, so the model reviews the exact text the server holds. MCP prompts
[
{
"content": {
"resource": {
"mimeType": "text/markdown",
"text": "# Retries\n\nA handler that raises is retried up to `max_retries` times, 3 by default, with exponential backoff. After the last attempt the job moves to the dead-letter queue. Make handlers idempotent: a retried job can run twice.\n",
"uri": "docs://pages/retries"
},
"type": "resource"
},
"role": "user"
},
{
"content": {
"text": "Review this docs page for accuracy. List each problem with the sentence it is in.",
"type": "text"
},
"role": "user"
}
]The prompts specification says a server should answer an unknown prompt name or a missing required argument with -32602. In mcp 2.3.0 both reach the client as -32603, so DocsServer overrides get_prompt: it checks the name and the required arguments against the public list_prompts(), then lets the SDK render the prompt. review_page itself raises MCPError with -32602 for a slug that has no page.
Prompt for a page that does not exist. review_page returns -32602; choose a slug from the index rather than asking the model to review a nonexistent page.
Prompt without its required argument. DocsServer.get_prompt returns -32602. Supply slug from the prompt’s published arguments.
Unknown prompt name. The override also returns -32602 for an unknown name. Discover prompts first. A plain MCPServer in this pinned SDK returns -32603 for this case and the missing argument; test_plain_server_reports_prompt_mistakes_as_internal_errors checks that difference so the override can be removed after an SDK fix.
6. Run every primitive and the tests
inspect_server.py connects the SDK client to the server object in the same process, calls each method once, including the failure cases, and saves what came back:
"""Use every tool, resource and prompt of docs_server.py and save what the client received.
Usage: python inspect_server.py [--write primitives-results.json]
The SDK client connects to the server object in the same process; the tests also run it over stdio.
"""
import argparse
import asyncio
import json
from pathlib import Path
from mcp import Client, MCPError
import docs_server
def dump(result):
"""The result as wire JSON, without the server identity block in _meta."""
data = result.model_dump(by_alias=True, exclude_none=True, mode="json")
data.pop("_meta", None)
return data
async def attempt(call):
try:
return {"result": dump(await call)}
except MCPError as error:
return {"error": {"code": error.error.code, "message": error.error.message}}
async def collect(server=docs_server.server):
async with Client(server) as client:
return {
"tools/list": dump(await client.list_tools()),
"resources/list": dump(await client.list_resources()),
"resources/templates/list": dump(await client.list_resource_templates()),
"prompts/list": dump(await client.list_prompts()),
"search worker": await attempt(client.call_tool("search_docs", {"query": "worker"})),
"search empty query": await attempt(client.call_tool("search_docs", {"query": " "})),
"read index": await attempt(client.read_resource("docs://index")),
"read page": await attempt(client.read_resource("docs://pages/retries")),
"read missing page": await attempt(client.read_resource("docs://pages/scheduling")),
"read unmatched URI": await attempt(client.read_resource("docs://drafts/retries")),
"get prompt": await attempt(client.get_prompt("review_page", {"slug": "retries"})),
"get prompt for a missing page": await attempt(client.get_prompt("review_page", {"slug": "scheduling"})),
"get prompt without slug": await attempt(client.get_prompt("review_page", {})),
"get unknown prompt": await attempt(client.get_prompt("rewrite_docs", {})),
}
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--write", type=Path, help="save the results to this JSON file")
results = asyncio.run(collect())
for name, value in results.items():
if name.endswith("/list"):
key = next(k for k in value if isinstance(value[k], list))
shown = f"{key}: " + ", ".join(item["name"] for item in value[key])
elif "error" in value:
shown = f"error {value['error']['code']}: {value['error']['message']}"
else:
shown = "isError=true" if value["result"].get("isError") else "ok"
print(f"{name:30} {shown}")
if (path := parser.parse_args(argv).write):
path.write_text(json.dumps(results, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(f"wrote {path}")
if __name__ == "__main__":
main().venv/bin/python inspect_server.py --write primitives-results.jsonHere is what a client receives from the finished server, recorded with inspect_server.py:
tools/list tools: search_docs
resources/list resources: docs_index
resources/templates/list resourceTemplates: docs_page
prompts/list prompts: review_page
search worker ok
search empty query isError=true
read index ok
read page ok
read missing page error -32602: No docs page named scheduling.
read unmatched URI error -32602: Unknown resource: docs://drafts/retries
get prompt ok
get prompt for a missing page error -32602: No docs page named scheduling.
get prompt without slug error -32602: Missing required arguments: slug
get unknown prompt error -32602: Unknown prompt: rewrite_docs
wrote primitives-results.jsonThe test suite reruns the same calls, validates structuredContent against the outputSchema, compares the run with the saved file and starts the server once as a real stdio subprocess:
.venv/bin/python -m unittest -v test_docs_server.pytest_server_runs_as_a_stdio_subprocess (test_docs_server.OverStdio.test_server_runs_as_a_stdio_subprocess) ... ok
test_plain_server_reports_prompt_mistakes_as_internal_errors (test_docs_server.PlainSdkDefault.test_plain_server_reports_prompt_mistakes_as_internal_errors)
Why DocsServer overrides get_prompt: mcp 2.3.0 answers these with -32603. ... ok
Tool 'search_docs' failed: "Error executing tool search_docs: The query is empty. Pass one or more words, such as 'worker timeout'."
Resource 'docs://pages/scheduling' failed: 'No docs page named scheduling.'
Resource 'docs://drafts/retries' failed: 'Unknown resource: docs://drafts/retries'
test_empty_query_is_a_tool_error (test_docs_server.Primitives.test_empty_query_is_a_tool_error) ... ok
test_lists_one_tool_resource_template_and_prompt (test_docs_server.Primitives.test_lists_one_tool_resource_template_and_prompt) ... ok
test_missing_page_is_invalid_params_not_empty_contents (test_docs_server.Primitives.test_missing_page_is_invalid_params_not_empty_contents) ... ok
test_prompt_embeds_the_page_and_uses_the_default_focus (test_docs_server.Primitives.test_prompt_embeds_the_page_and_uses_the_default_focus) ... ok
test_prompt_mistakes_are_invalid_params (test_docs_server.Primitives.test_prompt_mistakes_are_invalid_params) ... ok
test_saved_results_match_a_fresh_run (test_docs_server.Primitives.test_saved_results_match_a_fresh_run) ... ok
test_search_output_matches_its_output_schema (test_docs_server.Primitives.test_search_output_matches_its_output_schema) ... ok
test_template_reads_one_page_as_markdown (test_docs_server.Primitives.test_template_reads_one_page_as_markdown) ... ok
test_unmatched_uri_is_invalid_params (test_docs_server.Primitives.test_unmatched_uri_is_invalid_params) ... ok
----------------------------------------------------------------------
Ran 11 tests in 0.369s
OKAll 11 tests passed. The three lines in the middle are the SDK’s log messages for the expected failures, written to standard error. Every listing came back with ttlMs 0 and cacheScope private, the SDK defaults, which tell clients to treat each list as stale at once. MCP caching
Read the test source
"""Tests for docs_server.py: python -m unittest -v test_docs_server.py"""
import asyncio
import json
import sys
import unittest
from pathlib import Path
import jsonschema
from mcp import Client, MCPError, StdioServerParameters
from mcp.server import MCPServer
import inspect_server
HERE = Path(__file__).resolve().parent
class Primitives(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.r = asyncio.run(inspect_server.collect())
def test_lists_one_tool_resource_template_and_prompt(self):
r = self.r
self.assertEqual([t["name"] for t in r["tools/list"]["tools"]], ["search_docs"])
self.assertEqual([x["uri"] for x in r["resources/list"]["resources"]], ["docs://index"])
self.assertEqual([x["uriTemplate"] for x in r["resources/templates/list"]["resourceTemplates"]], ["docs://pages/{slug}"])
prompt = r["prompts/list"]["prompts"][0]
self.assertEqual([(a["name"], a.get("required", False)) for a in prompt["arguments"]], [("slug", True), ("focus", False)])
def test_search_output_matches_its_output_schema(self):
tool, result = self.r["tools/list"]["tools"][0], self.r["search worker"]["result"]
jsonschema.validate(result["structuredContent"], tool["outputSchema"])
self.assertEqual([hit["slug"] for hit in result["structuredContent"]["result"]], ["quickstart", "configuration"])
self.assertEqual(tool["annotations"], {"readOnlyHint": True, "openWorldHint": False})
def test_empty_query_is_a_tool_error(self):
result = self.r["search empty query"]["result"]
self.assertIs(result["isError"], True)
self.assertIn("The query is empty.", result["content"][0]["text"])
def test_template_reads_one_page_as_markdown(self):
content = self.r["read page"]["result"]["contents"][0]
self.assertEqual((content["uri"], content["mimeType"]), ("docs://pages/retries", "text/markdown"))
self.assertTrue(content["text"].startswith("# Retries"))
index = json.loads(self.r["read index"]["result"]["contents"][0]["text"])
self.assertEqual([page["slug"] for page in index], ["quickstart", "configuration", "retries"])
def test_missing_page_is_invalid_params_not_empty_contents(self):
self.assertEqual(self.r["read missing page"]["error"], {"code": -32602, "message": "No docs page named scheduling."})
def test_unmatched_uri_is_invalid_params(self):
self.assertEqual(self.r["read unmatched URI"]["error"]["code"], -32602)
def test_prompt_embeds_the_page_and_uses_the_default_focus(self):
first, second = self.r["get prompt"]["result"]["messages"]
self.assertEqual(first["content"]["type"], "resource")
self.assertEqual(first["content"]["resource"]["uri"], "docs://pages/retries")
self.assertEqual(second["content"]["text"], "Review this docs page for accuracy. List each problem with the sentence it is in.")
def test_prompt_mistakes_are_invalid_params(self):
self.assertEqual(self.r["get prompt without slug"]["error"], {"code": -32602, "message": "Missing required arguments: slug"})
self.assertEqual(self.r["get unknown prompt"]["error"]["code"], -32602)
self.assertEqual(self.r["get prompt for a missing page"]["error"]["code"], -32602)
def test_saved_results_match_a_fresh_run(self):
self.assertEqual(json.loads((HERE / "primitives-results.json").read_text()), self.r)
class PlainSdkDefault(unittest.TestCase):
def test_plain_server_reports_prompt_mistakes_as_internal_errors(self):
"""Why DocsServer overrides get_prompt: mcp 2.3.0 answers these with -32603."""
plain = MCPServer("plain")
@plain.prompt()
def greet(name: str) -> str:
return f"Say hello to {name}."
async def codes():
found = []
async with Client(plain) as client:
for name in ("greet", "nope"):
try:
await client.get_prompt(name, {})
except MCPError as error:
found.append(error.error.code)
return found
with self.assertLogs("mcp", level="ERROR"):
self.assertEqual(asyncio.run(codes()), [-32603, -32603])
class OverStdio(unittest.TestCase):
def test_server_runs_as_a_stdio_subprocess(self):
async def session():
params = StdioServerParameters(command=sys.executable, args=[str(HERE / "docs_server.py")])
async with Client(params) as client:
page = await client.read_resource("docs://pages/quickstart")
return page.contents[0].mime_type
self.assertEqual(asyncio.run(session()), "text/markdown")
if __name__ == "__main__":
unittest.main()Inspect resource and prompt messages
The wire protocol carries JSON requests and results, not Python function objects. These are the shapes it received in the recorded run: the template entry from resources/templates/list, a successful resources/read, and the error for a missing page.
{
"resources/templates/list": [
{
"description": "The Markdown text of one docs page.",
"mimeType": "text/markdown",
"name": "docs_page",
"title": "Docs page",
"uriTemplate": "docs://pages/{slug}"
}
],
"resources/read docs://pages/retries": {
"cacheScope": "private",
"contents": [
{
"mimeType": "text/markdown",
"text": "# Retries\n\nA handler that raises is retried up to `max_retries` times, 3 by default, with exponential backoff. After the last attempt the job moves to the dead-letter queue. Make handlers idempotent: a retried job can run twice.\n",
"uri": "docs://pages/retries"
}
],
"resultType": "complete",
"ttlMs": 0
},
"resources/read docs://pages/scheduling": {
"error": {
"code": -32602,
"message": "No docs page named scheduling."
}
}
}The SDK builds the template entry from the decorator: the URI pattern, the function name as name, the docstring as description, and the declared MIME type. When resources/read arrives, it first tries the fixed resources, then matches the URI against each template and calls the first function whose pattern fits, with the captured slug. If a URI matches neither a fixed resource nor a template, the SDK answers -32602 with Unknown resource. Errors are JSON-RPC errors here, not results with isError, because the application, not the model, asked for the resource and must handle the failure in code.
The prompt follows the same path. prompts/get validates the arguments against the function signature, calls review_page, and serializes each UserMessage into a role and a content block. The embedded resource carries its own uri and mimeType, so a host can show the reader which page the model saw.
The design choices
- Search returns URIs, not page text.
search_docscould return each full page. Returning slugs anddocs://URIs keeps the tool result small and lets the application decide which pages to read into the context, which matters when a search matches dozens of long pages. - A template, not a list of every page. Listing every page makes
resources/listgrow with the docs and go stale whenever a page is added. A template stays fixed; the cost is that a client cannot discover pages without the index resource or the search tool, which is why the server offers both. - Embed the page in the prompt, or link to it. Embedding sends the text held by this server at render time, reducing the risk of the host attaching an old local copy, but the whole page becomes model input tokens each time the host sends the prompt to a model.
prompts/getitself only retrieves the messages; no model runs until the host sends them. A text message that names the URI is shorter and leaves the read to the host, which may not do it.
Check your understanding
Read the explanation
The control hierarchy decides: the model calls tools, the user picks prompts and the application attaches resources.
Integrate MCP tools into ADK and LangGraph agents continues with the next part of the MCP application.
References
Checked October 21, 2026, against MCP specification 2026-07-28 and mcp 2.3.0.