L3 · 01 · Building MCP Servers (Python & Scala/JVM)
Level 2 taught you to use MCP servers. Now you build them. An MCP server is a small program that exposes tools, resources and prompts to any MCP-compatible host over a standard JSON-RPC protocol. This page builds one in Python with the official SDK, shows how the same thing works on the JVM from Scala, and covers testing, packaging and security.
By the end of this page you can
- Explain the MCP lifecycle: initialize, capability negotiation, tools/list, tools/call, shutdown.
- Build an MCP server in Python with
FastMCP: tools, resources and prompts. - Build the same server on the JVM (official Java SDK used from Scala) or plain JSON-RPC over stdio.
- Test a server with unit tests and the MCP Inspector, and package it for teammates.
- Apply the security rules: least-privilege tools, input validation, no ambient credentials.
What a server actually is
MCP is JSON-RPC 2.0 over a transport. The host starts your server as a subprocess (stdio) or connects over HTTP. After a handshake, the host lists what you offer and calls it. That's the whole protocol:
| You expose | Host sees | Example |
|---|---|---|
| Tools | Functions the model may call | check_inventory(sku, store) |
| Resources | Readable data, addressed by URI | policy://returns |
| Prompts | Reusable prompt templates | summarise_ticket |
A Python server with FastMCP
The official Python SDK's FastMCP turns typed functions into tools. Docstrings become tool descriptions, and type
hints become the input schema — so write them well (page L2·04 rules still apply).
try:
from mcp.server.fastmcp import FastMCP # mcp 1.x
except ImportError: # mcp 2.x renamed it
from mcp.server.mcpserver import MCPServer as FastMCP
mcp = FastMCP("store-tools")
INVENTORY = {
("EARBUDS-AIR2", "riverside"): 14,
("CASE-RUGGED", "riverside"): 0,
("CHARGER-65W", "riverside"): 32,
}
POLICIES = {"returns": "Unopened items: 30 days. Opened electronics: 14 days with receipt."}
@mcp.tool()
def check_inventory(sku: str, store: str) -> dict:
"""Check stock of a product at a store.
Args:
sku: product SKU, e.g. "EARBUDS-AIR2"
store: store slug, e.g. "riverside"
"""
qty = INVENTORY.get((sku.upper(), store.lower()))
if qty is None:
return {"found": False, "sku": sku, "store": store}
return {"found": True, "sku": sku, "store": store, "quantity": qty}
@mcp.tool()
def order_status(order_id: str) -> dict:
"""Look up an order's status. Returns found=False for unknown orders."""
demo = {"ORD-482913": {"status": "READY_FOR_PICKUP", "store": "riverside"}}
if order_id not in demo:
return {"found": False, "orderId": order_id}
return {"found": True, "orderId": order_id, **demo[order_id]}
@mcp.resource("policy://{name}")
def policy(name: str) -> str:
"""Store policy text by name (e.g. policy://returns)."""
return POLICIES.get(name, f"No policy named {name!r}")
@mcp.prompt()
def summarise_ticket(ticket: str) -> str:
"""One-line ticket summary for the daily report."""
return f"Summarise this support ticket in one line, keep order IDs verbatim:\n{ticket}"
if __name__ == "__main__":
mcp.run(transport="stdio")
Run it directly for a quick check, or connect a host:
python store_server.py # waits on stdio; Ctrl+C to quit npx @modelcontextprotocol/inspector python store_server.py # click-test tools in a browser
Unit-test the tools, not the transport
from store_server import check_inventory, order_status, policy
def test_known_sku_reports_quantity():
r = check_inventory("earbuds-air2", "Riverside")
assert r == {"found": True, "sku": "earbuds-air2", "store": "Riverside", "quantity": 14}
def test_unknown_sku_is_not_an_error():
assert check_inventory("NOPE", "riverside")["found"] is False
def test_unknown_order_is_not_an_error():
assert order_status("ORD-000000")["found"] is False
def test_missing_policy_says_so():
assert "No policy" in policy("warranty")
On the JVM: same protocol, two routes
For Scala/JVM shops there are two realistic options:
| Route | When |
|---|---|
Official MCP Java SDK (io.modelcontextprotocol.sdk) called from Scala, or Kotlin wrappers | Anything real: handles the lifecycle, schemas and transports for you |
| Plain JSON-RPC over stdio (below) | Learning the protocol, or a tiny internal tool with zero dependencies |
This minimal server answers initialize, tools/list and tools/call itself. It compiles with
plain scala-cli — no dependencies:
import scala.io.StdIn.readLine
import scala.util.matching.Regex
// Minimal stdio MCP server: enough protocol to expose one tool.
// Real servers should use the official Java SDK, which handles edge cases.
object MiniMcp:
private val inventory = Map("EARBUDS-AIR2" -> 14, "CHARGER-65W" -> 32)
private def strField(json: String, field: String): Option[String] =
new Regex(s"\"$field\"\\s*:\\s*\"([^\"]*)\"").findFirstMatchIn(json).map(_.group(1))
private def numField(json: String, field: String): Option[Long] =
new Regex(s"\"$field\"\\s*:\\s*(\\d+)").findFirstMatchIn(json).map(_.group(1).toLong)
private def reply(id: Long, resultJson: String): String =
s"""{"jsonrpc":"2.0","id":$id,"result":$resultJson}"""
private def toolResult(text: String): String =
s"""{"content":[{"type":"text","text":${toJsonString(text)}}]}"""
private def toJsonString(s: String): String =
"\"" + s.replace("\\", "\\\\").replace("\"", "\\\"") + "\""
def handle(line: String): Option[String] =
val id = numField(line, "id").getOrElse(-1L)
if line.contains("\"initialize\"") then
Some(reply(id, """{"protocolVersion":"2024-11-05","capabilities":{"tools":{}},""" +
""""serverInfo":{"name":"mini-inventory","version":"0.1.0"}}"""))
else if line.contains("notifications/") then None // notifications get no reply
else if line.contains("\"tools/list\"") then
Some(reply(id, """{"tools":[{"name":"check_inventory","description":"Stock of a SKU at the demo store",""" +
""""inputSchema":{"type":"object","properties":{"sku":{"type":"string"}},"required":["sku"]}}]}"""))
else if line.contains("\"tools/call\"") then
val qty = strField(line, "sku").map(s => inventory.getOrElse(s.toUpperCase, 0)).getOrElse(0)
Some(reply(id, toolResult(s"quantity=$qty")))
else if line.contains("\"ping\"") then Some(reply(id, "{}"))
else Some(s"""{"jsonrpc":"2.0","id":$id,"error":{"code":-32601,"message":"method not found"}}""")
def main(args: Array[String]): Unit =
Iterator.continually(readLine()).takeWhile(_ != null)
.flatMap(handle).foreach(println)
Paste an initialize JSON-RPC line at it and it answers. Point an IDE's MCP config at
scala-cli run MiniMcp.scala and the tool appears. The exercise: notice how much work the SDK saves you —
schema validation, every transport, error handling, spec versioning.
Packaging for teammates
- Pin versions (Python:
uv/pip-tools; JVM: build tool). A server that upgrades itself under a teammate is a debugging nightmare. - One-command start: the MCP config in the host should be copy-pasteable from your README, e.g.
uv run store_server.py. - Never bake in secrets. Read API keys from env vars; document which ones are needed. (L2·09 covers safe config.)
- Version the server in
serverInfo— it shows up in hosts and helps debug "which version are you running?". - Log to stderr, never stdout on stdio servers. stdout is the protocol channel; one stray
printcorrupts it.
Security review checklist
- Every tool argument is validated (types, enums, length limits) before use.
- Read-only by default; write tools need confirmation at the host layer.
- No tool returns secrets, tokens or raw stack traces to the model.
- Resource URIs can't escape their namespace (no
../../etc/passwdequivalent). - HTTP transports have real auth, not just an obscure URL.
- Run
store_server.pyunder the MCP Inspector and call each tool by hand. - Add a third tool
stores_near(postcode)with an allow-listed set of postcodes. Test it. - Connect the server to your IDE's MCP config and ask the assistant a question that needs it.
- Port one tool to
MiniMcp.scala. Then read the Java SDK's server example to see what it does differently.
Key takeaways
- MCP = JSON-RPC: initialize → list → call, over stdio or HTTP. SDKs do the plumbing.
- In Python,
FastMCPderives tool schemas from type hints and docstrings — write them as contracts. - Unit-test tool functions directly; use the Inspector for interactive checks.
- Least-privilege tools, validated arguments, secrets from env, logging on stderr.