Building an MCP Server and Client in Python: From 0 to 1 with Stdio and SSE Transports

MCP Server is a standardized interface that enables AI models to call external data sources and tools such as file systems, databases, or APIs, solving the inconsistencies of pre‑MCP Function Call formats and diverse API input/output schemas.

The article introduces MCP as a "USB‑C" for AI, unifying function call formats across major model providers and simplifying tool packaging. It then outlines the two supported transport protocols: Stdio for local execution and Server‑Sent Events (SSE) for cloud deployment.

Step 1 – Environment Configuration

Install the fast UV package (faster than pip) and initialize a new project:

curl -LsSf https://astral.sh/uv/install.sh | sh

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

uv init mcp-server

cd mcp-server

uv venv

source .venv/bin/activate   # Windows: .venv\Scripts\activate

uv add "mcp[cli]" httpx

Create the main server file:

touch main.py

Step 2 – Building Tool Functions

Define a mapping of documentation sites and implement asynchronous helpers to search the web via serper.dev and fetch page content:

docs_urls = {"langchain": "python.langchain.com/docs", "llama-index": "docs.llamaindex.ai/en/stable", "autogen": "microsoft.github.io/autogen/stable", "agno": "docs.agno.com", "openai-agents-sdk": "openai.github.io/openai-agents-python", "mcp-doc": "modelcontextprotocol.io", "camel-ai": "docs.camel-ai.org", "crew-ai": "docs.crewai.com"}

async def search_web(query: str) -> dict | None:
    payload = json.dumps({"q": query, "num": 3})
    headers = {"X-API-KEY": os.getenv("SERPER_API_KEY"), "Content-Type": "application/json"}
    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(SERPER_URL, headers=headers, data=payload, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except httpx.TimeoutException:
            return {"organic": []}

async def fetch_url(url: str):
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, timeout=30.0)
            soup = BeautifulSoup(response.text, "html.parser")
            return soup.get_text()
        except httpx.TimeoutException:
            return "Timeout error"

@mcp.tool()
async def get_docs(query: str, library: str):
    """Search the latest documentation for a given query and library.
    Supports langchain, llama-index, autogen, agno, openai‑agents‑sdk, mcp‑doc, camel‑ai, crew‑ai.
    """
    if library not in docs_urls:
        raise ValueError(f"Library {library} not supported by this tool")
    search_query = f"site:{docs_urls[library]} {query}"
    results = await search_web(search_query)
    if len(results["organic"]) == 0:
        return "No results found"
    text = ""
    for result in results["organic"]:
        text += await fetch_url(result["link"])
    return text

Step 3 – Stdio MCP Server (Local)

from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
import httpx, json, os
from bs4 import BeautifulSoup
from starlette.applications import Starlette
from mcp.server.sse import SseServerTransport
from starlette.requests import Request
from starlette.routing import Mount, Route
from mcp.server import Server
import uvicorn

load_dotenv()

mcp = FastMCP("Agentdocs")
USER_AGENT = "Agentdocs-app/1.0"
SERPER_URL = "https://google.serper.dev/search"
# docs_urls defined as above …
# search_web, fetch_url, get_docs defined as above …

if __name__ == "__main__":
    mcp.run(transport="stdio")

Run the server locally with:

uv run main.py

Step 3.2 – Client Configuration

For VS Code users, configure a Cline client by adding a mcpServers entry to the workspace settings:

{
  "mcpServers": {
    "mcp-server": {
      "command": "uv",
      "args": ["--directory", "
", "run", "main.py"]
    }
  }
}

For Cursor users, create a .cursor/mcp.json file with a similar command block.

Step 4 – SSE MCP Server (Remote)

from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
import httpx, json, os
from bs4 import BeautifulSoup
from starlette.applications import Starlette
from mcp.server.sse import SseServerTransport
from starlette.requests import Request
from starlette.routing import Mount, Route
from mcp.server import Server
import uvicorn

load_dotenv()

mcp = FastMCP("docs")
USER_AGENT = "docs-app/1.0"
SERPER_URL = "https://google.serper.dev/search"
# docs_urls, search_web, fetch_url, get_docs same as before

def create_starlette_app(mcp_server: Server, *, debug: bool = False) -> Starlette:
    sse = SseServerTransport("/messages/")
    async def handle_sse(request: Request) -> None:
        async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream):
            await mcp_server.run(read_stream, write_stream, mcp_server.create_initialization_options())
    return Starlette(debug=debug, routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ])

if __name__ == "__main__":
    mcp_server = mcp._mcp_server
    import argparse
    parser = argparse.ArgumentParser(description='Run MCP SSE‑based server')
    parser.add_argument('--host', default='0.0.0.0', help='Host to bind to')
    parser.add_argument('--port', type=int, default=8020, help='Port to listen on')
    args = parser.parse_args()
    starlette_app = create_starlette_app(mcp_server, debug=True)
    uvicorn.run(starlette_app, host=args.host, port=args.port)

Start the SSE server with:

uv run main.py --host 0.0.0.0 --port 8020

Step 4.2 – MCP Client (SSE)

import asyncio, json, os
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession
from mcp.client.sse import sse_client
from openai import AsyncOpenAI
from dotenv import load_dotenv

load_dotenv()

class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.openai = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL"))

    async def connect_to_sse_server(self, server_url: str):
        self._streams_context = sse_client(url=server_url)
        streams = await self._streams_context.__aenter__()
        self._session_context = ClientSession(*streams)
        self.session = await self._session_context.__aenter__()
        await self.session.initialize()
        print("Initialized SSE client...")
        print("Listing tools...")
        response = await self.session.list_tools()
        print("\nConnected to server with tools:", [tool.name for tool in response.tools])

    async def cleanup(self):
        if self._session_context:
            await self._session_context.__aexit__(None, None, None)
        if self._streams_context:
            await self._streams_context.__aexit__(None, None, None)

    async def process_query(self, query: str) -> str:
        messages = [{"role": "user", "content": query}]
        response = await self.session.list_tools()
        available_tools = [{"type": "function", "function": {"name": tool.name, "description": tool.description, "parameters": tool.inputSchema}} for tool in response.tools]
        completion = await self.openai.chat.completions.create(
            model=os.getenv("OPENAI_MODEL"), max_tokens=1000, messages=messages, tools=available_tools)
        assistant_message = completion.choices[0].message
        final_text = []
        if assistant_message.tool_calls:
            for tool_call in assistant_message.tool_calls:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)
                result = await self.session.call_tool(tool_name, tool_args)
                final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
                messages.extend([
                    {"role": "assistant", "content": None, "tool_calls": [tool_call]},
                    {"role": "tool", "tool_call_id": tool_call.id, "content": result.content[0].text}
                ])
                completion = await self.openai.chat.completions.create(
                    model=os.getenv("OPENAI_MODEL"), max_tokens=1000, messages=messages)
                if isinstance(completion.choices[0].message.content, (dict, list)):
                    final_text.append(str(completion.choices[0].message.content))
                else:
                    final_text.append(completion.choices[0].message.content)
        else:
            if isinstance(assistant_message.content, (dict, list)):
                final_text.append(str(assistant_message.content))
            else:
                final_text.append(assistant_message.content)
        return "\n".join(final_text)

    async def chat_loop(self):
        print("\nMCP Client Started!")
        print("Type your queries or 'quit' to exit.")
        while True:
            try:
                query = input("\nQuery: ").strip()
                if query.lower() == 'quit':
                    break
                response = await self.process_query(query)
                print("\n" + response)
            except Exception as e:
                print(f"\nError: {str(e)}")

async def main():
    if len(sys.argv) < 2:
        print("Usage: uv run client.py
")
        sys.exit(1)
    client = MCPClient()
    try:
        await client.connect_to_sse_server(server_url=sys.argv[1])
        await client.chat_loop()
    finally:
        await client.cleanup()

if __name__ == "__main__":
    import sys
    asyncio.run(main())

Run the client with:

uv run client.py http://0.0.0.0:8020/sse

The article concludes with screenshots of successful server and client logs, confirming that the MCP server can retrieve up‑to‑date documentation for various AI frameworks and that the client can query the server via both Stdio and SSE transports.