Build a Runnable MCP Loop in Python (stdio → streamable-http → LLM tool choice)

Attributed Chinese → English compile

Source: [MCP][02]快速入门MCP开发

Original author: 花酒锄作田 (Cnblogs) · Posted: 2025-09-15

This is an English rewrite of the original tutorial’s ideas and code patterns. It is not original work by the compiler. Always link the Chinese source; do not present this compile as the original.

Many MCP write-ups only show how to register a Server and paste it into Cursor. The Cnblogs post by 花酒锄作田 is useful for product engineers because it also builds the Client / Host side: list prompts, resources, and tools; call them over stdio; switch to streamable HTTP; then let an LLM decide which tool to invoke.

If you are shipping agents into a backend, that Client loop is the missing middle between “SDK demo” and “our service owns the tool session.” You need a reliable discover → bind → call → feed-back loop before you care which model sits on top.

Environment

The author used Python 3.13.5 (3.11+ is fine). Prefer uv or pip:

# uv
uv add mcp fastmcp

# or pip
python -m pip install mcp fastmcp

Notes from the source:

• The official mcp package ships FastMCP v1; community FastMCP has moved to v2—trying both while learning is fine.
• Write type hints, return types, and docstrings carefully. Those become the model-facing tool descriptions later.

Step 1 — Minimal FastMCP Server (stdio)

Concept: prompts, resources, and tools on one server with transport="stdio".

Illustrative shape (adapted [email protected]()def greet_user(name: str, style: str = "formal") -> str:
"""Greet a user with a specified style."""
if style == "friendly":
return f"Hey {name}! What's up?"
return f"Hello, {name}!"

@mcp.resource("greeting://{name}")def greeting_resource(name: str) -> str:
"""A simple greeting resource."""
return f"Hello, {name}!"

@mcp.resource("config://app")def get_config() -> str:
"""Static configuration data"""
return "App configuration here"

@mcp.tool()def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b

@mcp.tool()async def get_date() -> str:
"""Get today's date."""
return datetime.now().strftime("%Y-%m-%d")

@mcp.tool()async def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"It's always sunny in {city}!"

if name == "main":
mcp.run(transport="stdio")

**Preflight:** run the server script alone once. If imports fail, the Client will fail in a confusing way when it tries to spawn the child process.

> **Production caution — SSH / god-mode shell:** the original also demonstrates a remote SSH tool. Treat that as **high-risk**. Do not expose arbitrary remote command execution to a model without allowlists, authentication, and human confirmation. Prefer narrow, typed tools over “run anything on this host.”

## Step 2 — Stdio Client with `ClientSession`

The Client launches the Server as a subprocess via `StdioServerParameters` (absolute interpreter, script path, and cwd). Pattern from the source:
server_params = StdioServerParameters(
    command=str(Path(__file__).parent / ".venv" / "bin" / "python"),
    args=[str(Path(__file__).parent / "demo1-server.py")],
    cwd=str(Path(__file__).parent),
)

async def run():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            prompts = await session.list_prompts()
            print([p.name for p in prompts.prompts])

            tools = await session.list_tools()
            print([t.name for t in tools.tools])

            resource_content = await session.read_resource(AnyUrl("greeting://World"))
            block = resource_content.contents[0]
            if isinstance(block, types.TextResourceContents):
                print(block.text)

            result = await session.call_tool("add", arguments={"a": 5, "b": 3})
            print(result.content[0].text if result.content else result)
            print(result.structuredContent)

if __name__ == "__main__":
    asyncio.run(run())

Step 3 — Same Server over streamable-http

Server change:

mcp = FastMCP("custom", host="localhost", port=8001)

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Client change (conceptually): use streamablehttp_client("http://localhost:8001/mcp"), then the same ClientSession.initialize() / list_* / call_tool flow. The source notes a third return value, get_session_id, for session management—usually unused in hello-worlds.

This is the fork most product backends care about: stdio for desktop or host-local tools, HTTP for remotely deployed tool servers.

Docs and ecosystem starting points:

• MCP Python SDK / quickstart
• FastMCP docs (community) for transport options

Step 4 — Let the LLM choose tools

Server stays the same. Client:

1. Connect (HTTP example in the post).
2. Call list_tools() and mapThe original uses an OpenAI-compatible client pointed at Qwen / DashScope (qwen-plus, compatible-mode/v1). Any OpenAI-tools-compatible endpoint works the same way (DeepSeek, OpenAI, and similar).

Config sketch from the source’s supplementary modules:

{
  "llm": {
    "model": "qwen-plus",
    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
    "api_key": "your token"
  },
  "server": {
    "host": "127.0.0.1",
    "port": 8000
  }
}

Example interactive outcomes from the original session:

• “What is today’s date?” → get_date
• “Weather in Hefei?” → get_weather with {city: "合肥"}
• Numeric compare → custom comparison tool

That is the whole product loop in miniature: discover → bind schemas → model proposes → your code executes → feed results back.

Logging pitfall while integrating the LLM

The author’s sample logger can write to a file; if you stream-log, keep protocol traffic on the MCP pipes and human logs elsewhere. Mixing debug prints into a stdio Server’s stdout will break JSON-RPC—the same lesson every serious MCP guide repeats.

Why this matters for agents / MCP / RAG products

Shipping an LLM feature is less about a single chat completion and more about a reliable tool session: spawn or connect to servers, refresh schemas, bound the agent loop, and keep transports swappable (local stdio versus remote HTTP). The same Client you use for MCP tools is where you later hang RAG retrieval as a resource or tool—without rewriting the host when you add the next capability. Get this loop solid once, and every new tool becomes a schema change instead of a host rewrite.

Compiler (not original author)

English compile by YongBo Yu.

https://yongbo-yu.vercel.app · https://github.com/YongBoYu1

Original Chinese article © 花酒锄作田 / Cnblogs. Always link the source; do not present this compile as the original. each tool to an OpenAI-compatible function schema (name, description, parameters from inputSchema).

1. Call chat completions with tools=....
2. While the model emits tool calls: session.call_tool(name, args), append assistant and tool messages, call the model again.
3. Stop when there are no more tool calls.

Expected behavior (as reported in the original run): prompts listed, resource text returned, add yields 8 plus structured content.

Failure mode to remember: starting the Client also starts the Server. Server syntax or import errors look like Client connection failures—debug the Server first.

import asyncio
from pathlib import Path
from pydantic import AnyUrl

from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client

om the original). **Trim any SSH / remote-shell tools** before you run this locally unless you harden them first:

python
from datetime import datetime
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("custom")