P13 the custom tool builder python mcp protocol implementation

Project 13: “The Custom Tool Builder (Python)” — MCP Protocol Implementation

Attribute	Value
File	`KIRO_CLI_LEARNING_PROJECTS.md`
Main Programming Language	Python
Coolness Level	Level 4: Hardcore Tech Flex
Business Potential	5. Industry Disruptor (Ecosystem)
Difficulty	Level 3: Advanced
Knowledge Area	MCP Protocol Implementation

What you’ll build: A custom MCP server exposing fetch_stock_price(ticker).

Why it teaches Protocol: You learn MCP as JSON-RPC over stdio.

Success criteria:

Kiro calls your tool and parses real output.

Real World Outcome

You’ll have a working MCP server in Python that Kiro CLI can discover and use. When you configure it in your Kiro settings and ask “What’s the current price of AAPL?”, the following happens:

1. MCP Server Registration (in ~/.config/kiro/settings.json):

{
  "mcpServers": {
    "stock-prices": {
      "command": "python3",
      "args": ["/path/to/stock_mcp_server.py"],
      "env": {
        "STOCK_API_KEY": "your_api_key_here"
      }
    }
  }
}

2. Kiro CLI Session:

$ kiro
You: What's the current price of AAPL?

[Tool Call] stock-prices.fetch_stock_price(ticker="AAPL")

Tool Response:
{
  "ticker": "AAPL",
  "price": 178.42,
  "currency": "USD",
  "timestamp": "2025-01-02T14:32:00Z",
  "change": +2.15,
  "change_percent": +1.22
}

Kiro: Apple (AAPL) is currently trading at $178.42 USD, up $2.15 (+1.22%) from the previous close.

3. Server Logs (in your terminal running the MCP server):

[2025-01-02 14:32:00] INFO: MCP Server started on stdio
[2025-01-02 14:32:00] INFO: Registered tool: fetch_stock_price
[2025-01-02 14:32:15] INFO: Received JSON-RPC request: initialize
[2025-01-02 14:32:15] INFO: Sent capabilities: {tools: 1}
[2025-01-02 14:32:18] INFO: Tool called: fetch_stock_price(ticker="AAPL")
[2025-01-02 14:32:19] INFO: Fetching from Alpha Vantage API...
[2025-01-02 14:32:20] INFO: Response sent: {"price": 178.42, ...}

You’re seeing:

The MCP protocol handshake (initialize request/response)
Tool discovery (Kiro learns about fetch_stock_price)
JSON-RPC call serialization
Real-time API integration
Structured data flowing back to the LLM

This is the same pattern used by production MCP servers like @modelcontextprotocol/server-postgres, @modelcontextprotocol/server-github, and custom enterprise tools.

The Core Question You’re Answering

“How do I extend an AI coding agent with custom capabilities that go beyond its built-in tools?”

Before you write any code, think about this: LLMs are powerful, but they’re fundamentally text generators. They can’t fetch real-time stock prices, query proprietary databases, or interact with internal APIs. MCP bridges this gap by giving you a protocol to expose custom functionality as “tools” the LLM can call.

This project forces you to understand:

The JSON-RPC protocol - How requests and responses are structured
Stdio transport - Why MCP uses stdin/stdout instead of HTTP
Tool schemas - How to declare parameters and return types
Error handling - What happens when your tool fails
State management - How to maintain connection state across calls

By the end, you’ll see that MCP is just structured conversation: the LLM sends JSON requests, your server sends JSON responses, and Kiro orchestrates the exchange.

Concepts You Must Understand First

Stop and research these before coding:

JSON-RPC 2.0 Protocol
- What are the required fields in a JSON-RPC request? (jsonrpc, method, params, id)
- How do you distinguish a request from a response?
- What’s the difference between a notification and a request?
- How are errors represented in JSON-RPC?
- Book Reference: “Computer Networks, Fifth Edition” by Tanenbaum - Ch. 9 (Application Layer)
Stdio vs HTTP Transport
- Why does MCP use stdin/stdout instead of HTTP endpoints?
- How do you read JSON from stdin in Python without blocking?
- What’s the difference between line-buffered and unbuffered I/O?
- How do parent processes communicate with child processes?
- Book Reference: “Advanced Programming in the UNIX Environment, Third Edition” by Stevens - Ch. 15 (IPC)
Tool Schema Design
- How do you specify parameter types for an LLM? (JSON Schema)
- What’s the difference between required and optional parameters?
- How do you document what a tool does so the LLM uses it correctly?
- Why is return type structure important for LLM reasoning?
- Book Reference: “REST API Design Rulebook” by Mark Massé - Ch. 4 (Metadata Design)
Python Async I/O
- What’s the difference between sys.stdin.read() and sys.stdin.readline()?
- How do you flush stdout to ensure messages are sent immediately?
- Why might buffered output cause MCP protocol failures?
- How do you handle SIGTERM gracefully?

Questions to Guide Your Design

Before implementing, think through these:

Tool Registration
- How will you declare the fetch_stock_price tool to Kiro?
- What parameters does it need? (Just ticker, or also date, interval?)
- What should the return schema look like for maximum LLM usefulness?
- Should errors be returned as exceptions or structured error objects?
API Integration
- Which stock API will you use? (Alpha Vantage, Finnhub, Yahoo Finance?)
- How will you handle API rate limits?
- What happens if the API is down or slow?
- Should you cache responses to avoid redundant calls?
Error Handling
- What if the ticker symbol is invalid? (Return error or null?)
- What if the API key is missing or expired?
- What if the network request times out?
- How will you communicate these errors to the LLM clearly?
Protocol Compliance
- How will you implement the MCP handshake (initialize request)?
- What capabilities will you advertise? (Just tools, or also resources?)
- How will you parse incoming JSON-RPC without crashing on malformed input?
- What logging will help you debug protocol issues?

Thinking Exercise

Trace the MCP Handshake

Before coding, manually trace what happens when Kiro starts your MCP server:

Step 1: Kiro starts your server

$ python3 stock_mcp_server.py

Step 2: Kiro sends an initialize request via stdin:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "clientInfo": {"name": "kiro", "version": "1.0.0"}
  }
}

Step 3: Your server must respond with capabilities:

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": {}
    },
    "serverInfo": {"name": "stock-prices", "version": "0.1.0"}
  }
}

Step 4: Kiro sends a tools/list request:

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/list"
}

Step 5: Your server lists available tools:

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "tools": [
      {
        "name": "fetch_stock_price",
        "description": "Get real-time stock price for a ticker symbol",
        "inputSchema": {
          "type": "object",
          "properties": {
            "ticker": {
              "type": "string",
              "description": "Stock ticker symbol (e.g., AAPL, GOOGL)"
            }
          },
          "required": ["ticker"]
        }
      }
    ]
  }
}

Questions while tracing:

What happens if your server sends a response with the wrong id?
Why must you flush stdout after each JSON response?
What if Kiro sends a tools/call before you’ve responded to initialize?
How would you implement a timeout if the API takes too long?

The Interview Questions They’ll Ask

Prepare to answer these:

“Explain the difference between JSON-RPC and REST APIs. Why does MCP use JSON-RPC over stdio instead of HTTP?”
“Your MCP server is registered in Kiro settings, but the tool isn’t appearing. Walk me through your debugging process.”
“How would you handle API rate limits in an MCP server? Should you retry automatically or return an error to the LLM?”
“What happens if your MCP server crashes mid-conversation? How does Kiro detect this, and what’s the recovery process?”
“If you wanted to add authentication to your stock API calls (e.g., user-specific API keys), how would you design that in MCP?”
“Describe the lifecycle of an MCP server process. When is it started, and when is it terminated?”
“How would you test an MCP server without running Kiro? Can you simulate the protocol manually?”

Hints in Layers

Hint 1: Starting Point Your MCP server is just a Python script that reads JSON from stdin and writes JSON to stdout. Start by implementing a simple echo server: read a line, parse it as JSON, send back a response with the same id. Once that works, add the MCP-specific methods (initialize, tools/list, tools/call).

Hint 2: Structure Create a handle_request(request) function that dispatches based on request["method"]. Use a dictionary to map methods to handler functions:

handlers = {
    "initialize": handle_initialize,
    "tools/list": handle_tools_list,
    "tools/call": handle_tools_call
}

Hint 3: JSON-RPC Response Format Every response must include:

"jsonrpc": "2.0"
"id": <same as request>
Either "result": {...} for success or "error": {...} for failure

Always flush stdout after writing: sys.stdout.flush()

Hint 4: Tool Schema For tools/list, return a list of tools with JSON Schema for parameters. The inputSchema must follow JSON Schema Draft 7:

{
  "name": "fetch_stock_price",
  "description": "Get current stock price",
  "inputSchema": {
    "type": "object",
    "properties": {
      "ticker": {"type": "string", "description": "e.g., AAPL"}
    },
    "required": ["ticker"]
  }
}

Hint 5: API Integration For the stock API, use requests with error handling:

try:
    response = requests.get(api_url, params={"symbol": ticker}, timeout=5)
    response.raise_for_status()
    data = response.json()
except requests.exceptions.Timeout:
    return {"error": "API timeout"}
except requests.exceptions.RequestException as e:
    return {"error": f"API error: {str(e)}"}

Hint 6: Debugging Log everything to stderr (not stdout, which is used for protocol):

import sys
sys.stderr.write(f"[DEBUG] Received request: {request}\n")
sys.stderr.flush()

Run your server manually and paste JSON requests to test:

$ python3 stock_mcp_server.py
{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}

Books That Will Help

Topic	Book	Chapter
JSON-RPC Protocol	“Computer Networks, Fifth Edition” by Tanenbaum	Ch. 9 (RPC and Middleware)
Stdio/Process Communication	“Advanced Programming in the UNIX Environment” by Stevens	Ch. 15 (IPC)
JSON Schema	“REST API Design Rulebook” by Mark Massé	Ch. 4 (Metadata Design)
Python I/O	“Fluent Python, 2nd Edition” by Luciano Ramalho	Ch. 21 (Asynchronous Programming)
API Design	“REST API Design Rulebook” by Mark Massé	Ch. 2 (Identifier Design)

Common Pitfalls & Debugging

Problem 1: “Kiro doesn’t see my MCP server”

Why: The command path in settings.json is incorrect, or Python isn’t in PATH
Fix: Use absolute paths: "command": "/usr/bin/python3" and "args": ["/full/path/to/server.py"]
Quick test: Run the exact command manually: /usr/bin/python3 /full/path/to/server.py

Problem 2: “Server starts but tool doesn’t appear”

Why: You didn’t respond to tools/list correctly, or JSON is malformed
Fix: Add logging to stderr and check that tools/list returns valid JSON Schema

Quick test: Pipe a tools/list request manually:

$ echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | python3 server.py

Problem 3: “Tool calls return no data”

Why: You’re not flushing stdout, or the API call is failing silently
Fix: Always sys.stdout.flush() after writing JSON, and log API errors to stderr
Quick test: Add sys.stderr.write(f"API response: {data}\n") before returning

Problem 4: “Server crashes on malformed JSON”

Why: Kiro sent unexpected input, or your JSON parsing is brittle

Fix: Wrap json.loads() in try/except and return a JSON-RPC error:

try:
  request = json.loads(line)
except json.JSONDecodeError as e:
  error_response = {
      "jsonrpc": "2.0",
      "id": None,
      "error": {"code": -32700, "message": "Parse error"}
  }

Quick test: Send invalid JSON to your server and verify it doesn’t crash

Problem 5: “API rate limit exceeded”

Why: You’re calling the API on every request without caching

Fix: Implement a simple cache with TTL:

cache = {}
def fetch_with_cache(ticker):
  if ticker in cache and time.time() - cache[ticker]["time"] < 60:
      return cache[ticker]["data"]
  data = fetch_from_api(ticker)
  cache[ticker] = {"data": data, "time": time.time()}
  return data

Quick test: Call the same ticker twice rapidly and verify only one API call is made