The Complete Guide to Implementing the Model Context Protocol in 2025: From Spec Evolution to Streamable HTTP

Hello!
MCP is spreading through the LLM industry at breakneck speed, and today we want to focus specifically on the implementation side.

We say "MCP" as if it were a single thing, but the specification people call the "standard" has actually changed quite a bit in a short period. So we will walk through the variations of the spec in order, and then build a real implementation.

The Model Context Protocol (MCP), announced by Anthropic in late 2024, marked an important turning point for the AI field.

It standardized the tool-invocation capability (known as tool use) that each AI vendor had previously implemented in its own way, providing a unified mechanism for connecting AI models with external systems

In this article, we trace in detail the technical evolution of MCP from its birth to the present, and explain the best implementation approach as of 2025, complete with full source code. In particular, from the perspective of implementers who tend to get whipsawed by spec changes, we clarify why things converged on the current form and which implementation approach you should take going forward.

Chapter 1: The Problem MCP Set Out to Solve

The chaotic state of AI tool invocation

Until 2024, each model provider used its own tool invocation specification. OpenAI had Function Calling, Anthropic and Meta had Tool Use, Google had Function Declarations — the names and the implementations were all over the place.

Even for something as simple as "get the current time," developers had to prepare a different implementation for each LLM provider.

Granted, when a new technology first emerges, putting interoperability on the back burner is a classic pattern in the software industry.

But this situation was not merely tedious — it limited what AI agents could become. Building systems that combine multiple AI models, or applications that can swap models, was difficult, and tool reusability was remarkably poor.

MCP as the answer

MCP was a clear answer to this chaos. Built on JSON-RPC 2.0, it standardized everything needed for AI-tool integration, from tool definitions to session management and streaming communication. Implement an MCP server once, and the same tools become usable from Claude Desktop, VS Code, and any other MCP-capable client.

Chapter 2: How the Spec Evolved, and Why

The stdio era — an ideal design for local execution

Early MCP assumed communication over standard input/output (stdio). This is the most basic form of inter-process communication, and in local environments it was efficient.

# Conceptual early stdio implementation
import sys
import json

while True:
    line = sys.stdin.readline()
    request = json.loads(line)
    # Handle the request
    response = handle_request(request)
    sys.stdout.write(json.dumps(response) + "\n")
    sys.stdout.flush()

However, this design has a fundamental constraint.

That's right — stdio is a communication model premised on a parent-child process relationship, and it is inherently unsuited to communication over a network.

One wonders how many engineers today even think of stdio as a communication channel. Aside from people like me who wrote code back when a black console was all we had, or the folks still writing C for a living, stdio is probably not something most people consciously think about.

Still, for developers using it from VS Code or a local Claude Desktop, it posed no particular problem.

But serving multiple concurrent users as a web service, or running as part of a distributed system, was simply never part of the design ("why not?", you might ask).

The HTTP+SSE era — the first step toward network support

Next came the approach combining HTTP POST with Server-Sent Events (SSE). It is an asymmetric model: HTTP POST from client to server, SSE from server to client.

This approach had a certain logic to it. SSE is natively supported in browsers, and real-time progress notifications were possible. But implementation complexity grew. Session management, separate endpoints, connection synchronization — there was simply too much for developers to keep track of.
That is the kind of tedium peculiar to SSE.

The Streamable HTTP era — a mature, unified specification

And now, as of 2025, MCP has converged on Streamable HTTP: a refined design in which a single HTTP endpoint handles both regular request/response and streaming.

# The basic idea of Streamable HTTP
@app.post("/mcp")
async def mcp_endpoint(request: Request):
    if "text/event-stream" in request.headers.get("accept", ""):
        # Streaming response
        return StreamingResponse(generate_sse())
    else:
        # Regular JSON response
        return JSONResponse(handle_request(await request.json()))

This unification greatly reduced implementation complexity while improving scalability at the same time. Laying out the characteristics of each transport reveals the following progression.

The evolution of MCP transports

What this table makes clear is that MCP started with stdio, specialized for local execution, passed through the transitional HTTP+SSE implementation to answer the demand for remote access, and finally converged on Streamable HTTP as the unified solution.

Today there is a clear division of use: stdio for local-only scenarios, Streamable HTTP whenever remote access is a possibility.

The essential significance of streaming

Now — why is streaming such a good thing in MCP in the first place?

What is the point of returning responses in streaming form?

Let's start there.

"Streaming" carries a meaning that goes well beyond a mere choice of transport.

First, let's compare the traditional synchronous communication model with the asynchronous streaming model in diagrams.

The synchronous model looks something like this.

[Traditional synchronous model]

User ──────> Web UI ──────> LLM ──────> MCP Server
                               │             │
                               │             ▼
                               │        [Execute task]
                               │         (long-running)
                               │             │
                               ◄─────────────┘
                               │         (waiting...)
                               ▼
                       [Generate response]
                               │
User ◄────── Web UI ◄──────────┘

Problem: the LLM waits blindly until the task completes

Next, the streaming model

[Streaming model]

User ──────> Web UI ──────> LLM ──────> MCP Server
     │          │             │             │
     │          │             │             ▼
     │          │             │        [Task started]
     │          │             │             │
     │          │             ◄─ stream ────┤ "Task started"
     │          │             │             │
     │          │       [Interim reply]     ▼
     │          ◄─ stream ────┤        [Processing 30%]
     │      "Working..."      │             │
     ◄──────────┤             ◄─ stream ────┤ "30% done"
                │             │             │
                │      [Adjust strategy]    ▼
                │             │        [Processing 60%]
                │             ◄─ stream ────┤ "60% done"
     │          ◄─ stream ────┤             │
     │      "Almost there..." │             ▼
     ◄──────────┤             │        [Task complete]
                │             ◄─ stream ────┤ "Result: XXX"
                │             │             │
                │        [Final reply]      │
                ◄─ stream ────┤             │
     │      "All done"        │             │
     ◄──────────┘             │             │

Benefit: every participant shares the state in real time

In application development centered on AI models, for the model to make progress step by step while conversing with external tools, a mechanism that returns intermediate progress as it happens is indispensable.

The AI needs to keep generating its response while understanding "what is happening right now" and "how far the work has progressed." A synchronous request model that returns only the final result in one lump usually cannot satisfy that requirement.

[Adaptive processing via streaming]

LLM ──────> MCP Server (database search)
 │              │
 │              ▼ stream: "Searched 100 of 1000 records..."
 │◄─────────────┤
 │              │
 ├─Decision: taking longer than expected
 │       
 ├──────────> MCP Server (narrow the search criteria)
 │              │
 │              ▼ stream: "Criteria changed: narrowed to 50 records"
 │◄─────────────┤
 │              │
 │              ▼ stream: "Search complete"
 │◄─────────────┤
 │
 ▼
[Generate response for the user]
"I optimized the search scope and retrieved results
 from the 50 most relevant records"

The greatest significance of streaming is that the model no longer has to keep "waiting" on external work — it can "continue generating its response while observing the progress of that work." AI models have no true asynchronous processing of their own. When an external tool ran for a long time, the model had no choice but to sit blocked on the response. With streaming, the external MCP server returns partial progress and intermediate data to the AI as it goes, and the AI can make its next decision each time something arrives.

[Improved user experience]

Before:
User ──> Web UI ──> [Loading spinner...] ──> (anxious waiting) ──> result

Streaming:
User ──> Web UI ┬─> "Connecting to the database..."
                ├─> "Searching 1,000 records..."
                ├─> "Analyzing related data..."
                ├─> "Formatting results..."
                └─> "Done: [detailed results]"

The user can see what is happening at each step,
greatly improving transparency and trust

Streaming responses are also decisively important for the application. In a conventional server implementation, when a task took a long time, the user was left in an opaque "wait for the response" state. In an AI-centered UI, that translates directly into the quality of the experience. When the MCP server streams results and logs incrementally, the application can reflect them in the UI as they arrive, making "what the AI is doing" visible.

For example, when an external tool is rendering a video, sending progress like "67% complete" back to the web UI clearly improves the UX.

Moreover, this mechanism ties directly into making AI agents more sophisticated. Having external results flow in as a stream makes it possible, for the first time, for the AI model to "look at partial results and change its strategy". That is what makes this feel so distinctly like the AI era.

[Recovering from errors]

LLM ──────> MCP Server (API call)
 │              │
 │              ▼ stream: "Connecting to the API..."
 │◄─────────────┤
 │              │
 │              ▼ stream: "Error: rate limited"
 │◄─────────────┤
 │              
 ├─Decision: use a fallback
 │       
 ├──────────> MCP Server (fetch from cache)
 │              │
 │              ▼ stream: "Fetching from cache..."
 │◄─────────────┤
 │              │
 │              ▼ stream: "Success: data retrieved"
 │◄─────────────┤
 │
 ▼
[Response to the user]
"Fetching the latest data failed, so I used
 cached data from one hour ago"

If an anomaly is detected midway through a long task, the AI can immediately propose an alternative, or recognize the moments when additional instructions are needed after work has begun. With a synchronous response returned all at once, this kind of incremental decision-making was fundamentally impossible.

In short, the significance of MCP servers returning streaming responses is not merely a matter of efficiency. It is the foundation for a new execution model in which the AI model, the MCP server, and the application share the same timeline and "work together while conversing" through incremental exchanges of information. And Streamable HTTP can be called MCP's mature form: it maximizes the value of streaming while keeping implementation complexity to a minimum.

Chapter 3: A Complete Implementation — a Spec-Compliant Server with FastMCP

Why you should use FastMCP

The MCP spec looks simple at first glance, but once you try to implement a compliant server, the number of details to handle balloons: batch request handling, 204 responses for notification-only messages, managing the Mcp-Session-Id header, and much more.

FastMCP absorbs all of this and gives developers an environment where they can focus on business logic.

A complete MCP server implementation

Below is a production-ready MCP server implementation using FastMCP.

#!/usr/bin/env python3
"""
Production-Ready MCP DateTime Server
A complete MCP server implementation ready for production use
"""
import asyncio
from datetime import datetime, timedelta
from typing import Dict, Any, Optional
import pytz
import logging
from mcp.server.fastmcp import FastMCP

# Logging configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
logger = logging.getLogger("mcp-datetime-server")

# Timezone configuration
JST = pytz.timezone('Asia/Tokyo')

# Create the FastMCP instance
mcp = FastMCP(
    name="datetime-server-production",
    version="1.0.0"
)

# Server metadata
mcp.metadata = {
    "description": "Production-ready datetime server with JST support",
    "author": "Your Organization",
    "timezone": "Asia/Tokyo",
    "supported_languages": ["ja", "en"]
}


@mcp.tool(
    description="Gets the current time in Japan Standard Time (JST)"
)
async def get_current_time(format: str = "standard") -> str:
    """
    Return the current time in the specified format
    
    Args:
        format: output format (standard, iso, unix)
    """
    now = datetime.now(JST)
    
    if format == "iso":
        return now.isoformat()
    elif format == "unix":
        return str(int(now.timestamp()))
    else:
        return f"The current time is {now.strftime('%H:%M:%S')}"


@mcp.tool(
    description="Gets today's date in Japanese format"
)
async def get_current_date(
    include_era: bool = False
) -> str:
    """
    Return the current date (output uses Japanese date formatting)
    
    Args:
        include_era: whether to include the Japanese era (wareki)
    """
    now = datetime.now(JST)
    date_str = now.strftime("%Y年%m月%d日")
    day_of_week = ["月", "火", "水", "木", "金", "土", "日"][now.weekday()]
    
    result = f"今日は {date_str} ({day_of_week}曜日) です"  # "Today is <date> (<weekday>)" in Japanese
    
    if include_era:
        # Reiwa era calculation (the era started May 1, 2019)
        reiwa_start = datetime(2019, 5, 1, tzinfo=JST)
        if now >= reiwa_start:
            reiwa_year = now.year - 2018
            result += f" (令和{reiwa_year}年)"  # "(Reiwa year N)"
    
    return result


@mcp.tool(
    description="Calculates the date a given number of days ahead"
)
async def calculate_future_date(
    days: int,
    from_date: Optional[str] = None
) -> str:
    """
    Calculate a future or past date (output uses Japanese date formatting)
    
    Args:
        days: number of days (negative for the past)
        from_date: starting date (ISO format; today if omitted)
    """
    if from_date:
        base_date = datetime.fromisoformat(from_date).replace(tzinfo=JST)
    else:
        base_date = datetime.now(JST)
    
    target_date = base_date + timedelta(days=days)
    date_str = target_date.strftime("%Y年%m月%d日")
    day_of_week = ["月", "火", "水", "木", "金", "土", "日"][target_date.weekday()]
    
    if days > 0:
        return f"{days}日後は {date_str} ({day_of_week}曜日) です"  # "N days from now is <date>"
    elif days < 0:
        return f"{abs(days)}日前は {date_str} ({day_of_week}曜日) でした"  # "N days ago was <date>"
    else:
        return f"指定日は {date_str} ({day_of_week}曜日) です"  # "The specified day is <date>"


@mcp.tool(
    description="Calculates the number of days between two dates"
)
async def calculate_days_between(
    date1: str,
    date2: str
) -> str:
    """
    Calculate the difference in days between two dates
    
    Args:
        date1: the first date (ISO format)
        date2: the second date (ISO format)
    """
    d1 = datetime.fromisoformat(date1).replace(tzinfo=JST)
    d2 = datetime.fromisoformat(date2).replace(tzinfo=JST)
    
    diff = abs((d2 - d1).days)
    
    if d1 < d2:
        return f"It is {diff} days from {date1} to {date2}"
    elif d1 > d2:
        return f"It is {diff} days from {date2} to {date1}"
    else:
        return "The dates are the same"


@mcp.resource(
    uri="timezone://current",
    name="Current Timezone Information",
    description="Detailed information about the current timezone",
    mime_type="application/json"
)
async def get_timezone_resource() -> Dict[str, Any]:
    """Provide timezone information as a resource"""
    now = datetime.now(JST)
    utc_now = datetime.now(pytz.UTC)
    
    return {
        "timezone": "Asia/Tokyo",
        "abbreviation": "JST",
        "offset": "+09:00",
        "offset_seconds": 32400,
        "current_jst": now.isoformat(),
        "current_utc": utc_now.isoformat(),
        "is_dst": False
    }


@mcp.on_initialize
async def on_initialize(params: Dict[str, Any]) -> None:
    """Handling at server initialization"""
    client_info = params.get("clientInfo", {})
    logger.info(f"MCP Server initialized by {client_info.get('name', 'unknown')}")
    logger.info(f"Protocol version: {params.get('protocolVersion')}")


@mcp.on_error
async def on_error(error: Exception) -> None:
    """Handling when an error occurs"""
    logger.error(f"MCP Server error: {error}", exc_info=True)


def main():
    """Main entry point"""
    import argparse
    
    parser = argparse.ArgumentParser(
        description="Production MCP DateTime Server"
    )
    parser.add_argument(
        "--port", type=int, default=8080,
        help="Server port (default: 8080)"
    )
    parser.add_argument(
        "--host", type=str, default="0.0.0.0",
        help="Server host (default: 0.0.0.0)"
    )
    
    args = parser.parse_args()
    
    print(f"Starting MCP Server on {args.host}:{args.port}")
    print(f"Endpoint: http://{args.host}:{args.port}/mcp")
    
    # FastMCP automates all of the protocol handling
    mcp.run(
        transport="streamable-http",
        host=args.host,
        port=args.port
    )


if __name__ == "__main__":
    main()

The important point about this code is that using FastMCP's decorators automatically guarantees spec compliance. Developers can focus on tool logic without having to think about protocol details.

Chapter 4: Client Implementations — Integrating with Each LLM Provider

With the MCP server complete, the next step is the client side. We need to integrate each LLM provider's API with MCP.

Claude (Anthropic) client implementation

Claude, being the model from Anthropic — MCP's creator — integrates with MCP the most naturally.

import asyncio
from typing import List, Dict, Any
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
from anthropic import AsyncAnthropic

class ClaudeMCPIntegration:
    def __init__(self, mcp_url: str, api_key: str):
        self.mcp_url = mcp_url
        self.anthropic = AsyncAnthropic(api_key=api_key)
        self.session = None
        
    async def connect(self):
        """Connect to the MCP server"""
        transport = await sse_client(self.mcp_url)
        self.session = ClientSession(
            transport.read_stream,
            transport.write_stream
        )
        await self.session.__aenter__()
        await self.session.initialize()
        
    async def ask_claude_with_tools(self, prompt: str) -> str:
        """Ask Claude a question using MCP tools"""
        # Get the list of MCP tools
        tools_result = await self.session.list_tools()
        
        # Convert to the Claude API format
        claude_tools = []
        for tool in tools_result.tools:
            claude_tools.append({
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema
            })
        
        # Query Claude
        response = await self.anthropic.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            tools=claude_tools,
            messages=[{"role": "user", "content": prompt}]
        )
        
        # Handle tool calls
        for content in response.content:
            if content.type == "tool_use":
                # Execute the MCP tool
                result = await self.session.call_tool(
                    content.name,
                    content.input or {}
                )
                
                # Return the result to Claude to generate the final answer
                final_response = await self.anthropic.messages.create(
                    model="claude-3-5-sonnet-20241022",
                    max_tokens=1024,
                    messages=[
                        {"role": "user", "content": prompt},
                        {"role": "assistant", "content": response.content},
                        {
                            "role": "user",
                            "content": [{
                                "type": "tool_result",
                                "tool_use_id": content.id,
                                "content": result.content[0].text
                            }]
                        }
                    ]
                )
                
                return final_response.content[0].text
        
        return response.content[0].text

OpenAI GPT client implementation

OpenAI's Function Calling also integrates seamlessly with MCP tools.

from openai import AsyncOpenAI
import json

class OpenAIMCPIntegration:
    def __init__(self, mcp_url: str, api_key: str):
        self.mcp_url = mcp_url
        self.openai = AsyncOpenAI(api_key=api_key)
        self.session = None
    
    async def connect(self):
        """Connect to the MCP server"""
        transport = await sse_client(self.mcp_url)
        self.session = ClientSession(
            transport.read_stream,
            transport.write_stream
        )
        await self.session.__aenter__()
        await self.session.initialize()
    
    async def ask_gpt_with_tools(self, prompt: str) -> str:
        """Ask GPT a question using MCP tools"""
        # Get the list of MCP tools
        tools_result = await self.session.list_tools()
        
        # Convert to the OpenAI format
        openai_tools = []
        for tool in tools_result.tools:
            openai_tools.append({
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema
                }
            })
        
        # Query GPT
        response = await self.openai.chat.completions.create(
            model="gpt-4-turbo-preview",
            tools=openai_tools,
            messages=[{"role": "user", "content": prompt}]
        )
        
        message = response.choices[0].message
        
        # Handle tool calls
        if message.tool_calls:
            tool_results = []
            for tool_call in message.tool_calls:
                # Execute the MCP tool
                result = await self.session.call_tool(
                    tool_call.function.name,
                    json.loads(tool_call.function.arguments)
                )
                
                tool_results.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result.content[0].text
                })
            
            # Generate the final answer
            final_response = await self.openai.chat.completions.create(
                model="gpt-4-turbo-preview",
                messages=[
                    {"role": "user", "content": prompt},
                    message,
                    *tool_results
                ]
            )
            
            return final_response.choices[0].message.content
        
        return message.content

Google Gemini client implementation

Gemini can be implemented with the same pattern.

import google.generativeai as genai

class GeminiMCPIntegration:
    def __init__(self, mcp_url: str, api_key: str):
        self.mcp_url = mcp_url
        genai.configure(api_key=api_key)
        self.session = None
        self.model = None
    
    async def connect(self):
        """Connect to the MCP server"""
        transport = await sse_client(self.mcp_url)
        self.session = ClientSession(
            transport.read_stream,
            transport.write_stream
        )
        await self.session.__aenter__()
        await self.session.initialize()
        
        # Get tool definitions and initialize the model
        tools_result = await self.session.list_tools()
        
        functions = []
        for tool in tools_result.tools:
            functions.append({
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema
            })
        
        self.model = genai.GenerativeModel(
            model_name="gemini-1.5-pro",
            tools=functions
        )
    
    async def ask_gemini_with_tools(self, prompt: str) -> str:
        """Ask Gemini a question using MCP tools"""
        response = self.model.generate_content(prompt)
        
        # Handle function calls
        for part in response.candidates[0].content.parts:
            if hasattr(part, 'function_call'):
                fc = part.function_call
                
                # Execute the MCP tool
                result = await self.session.call_tool(
                    fc.name,
                    dict(fc.args)
                )
                
                # Return the result to generate the final answer
                function_response = genai.protos.FunctionResponse(
                    name=fc.name,
                    response={"result": result.content[0].text}
                )
                
                final_response = self.model.generate_content(
                    contents=[
                        {"role": "user", "parts": [{"text": prompt}]},
                        {"role": "model", "parts": response.candidates[0].content.parts},
                        {"role": "user", "parts": [{"function_response": function_response}]}
                    ]
                )
                
                return final_response.candidates[0].content.parts[0].text
        
        return response.candidates[0].content.parts[0].text

Chapter 5: Important Implementation Considerations

Error handling and retries

In production, you need to cope with transient network problems and server overload.

import asyncio
from typing import TypeVar, Callable
import random

T = TypeVar('T')

async def retry_with_exponential_backoff(
    func: Callable[[], T],
    max_retries: int = 3,
    base_delay: float = 1.0
) -> T:
    """Retry implementation with exponential backoff"""
    for attempt in range(max_retries):
        try:
            return await func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            logger.warning(
                f"Attempt {attempt + 1} failed: {e}. "
                f"Retrying in {delay:.2f} seconds..."
            )
            await asyncio.sleep(delay)

Session management and connection pooling

For long-running services, proper session management is important.

class MCPConnectionPool:
    """Connection pool manager for the MCP server"""
    
    def __init__(self, mcp_url: str, max_connections: int = 10):
        self.mcp_url = mcp_url
        self.max_connections = max_connections
        self.connections = asyncio.Queue(maxsize=max_connections)
        self.lock = asyncio.Lock()
    
    async def acquire(self) -> ClientSession:
        """Acquire a connection"""
        try:
            return self.connections.get_nowait()
        except asyncio.QueueEmpty:
            async with self.lock:
                if self.connections.qsize() < self.max_connections:
                    # Create a new connection
                    transport = await sse_client(self.mcp_url)
                    session = ClientSession(
                        transport.read_stream,
                        transport.write_stream
                    )
                    await session.__aenter__()
                    await session.initialize()
                    return session
                else:
                    # Wait until a connection frees up
                    return await self.connections.get()
    
    async def release(self, session: ClientSession):
        """Release a connection"""
        await self.connections.put(session)

Performance optimization

For large-scale deployments, performance optimization matters.

import functools
from typing import Any, Dict
import hashlib
import pickle

class MCPResponseCache:
    """Cache implementation for MCP responses"""
    
    def __init__(self, ttl_seconds: int = 300):
        self.cache: Dict[str, tuple[Any, float]] = {}
        self.ttl_seconds = ttl_seconds
    
    def _generate_key(self, tool_name: str, arguments: Dict[str, Any]) -> str:
        """Generate a cache key"""
        data = f"{tool_name}:{pickle.dumps(arguments, protocol=pickle.HIGHEST_PROTOCOL)}"
        return hashlib.sha256(data.encode()).hexdigest()
    
    async def get_or_fetch(
        self,
        tool_name: str,
        arguments: Dict[str, Any],
        fetch_func: Callable
    ) -> Any:
        """Get from cache, or fetch if missing"""
        key = self._generate_key(tool_name, arguments)
        
        # Check the cache
        if key in self.cache:
            result, timestamp = self.cache[key]
            if time.time() - timestamp < self.ttl_seconds:
                return result
        
        # Fetch and cache the result
        result = await fetch_func()
        self.cache[key] = (result, time.time())
        return result

Chapter 6: Operations and Monitoring

This chapter is half a bonus, but we include it as a reminder of what not to forget when building an MCP server.

Implementing health checks

In production you need to monitor service health continuously, so at the very least add a health check endpoint

@mcp.tool(
    description="Checks the server's health"
)
async def health_check() -> str:
    """Health check tool"""
    checks = {
        "server_status": "healthy",
        "timezone_check": datetime.now(JST).isoformat(),
        "memory_usage": get_memory_usage(),
        "uptime": get_uptime()
    }
    
    return json.dumps(checks, ensure_ascii=False, indent=2)

def get_memory_usage() -> str:
    """Get memory usage"""
    import psutil
    process = psutil.Process()
    memory_mb = process.memory_info().rss / 1024 / 1024
    return f"{memory_mb:.2f} MB"

def get_uptime() -> str:
    """Get uptime"""
    global server_start_time
    if not server_start_time:
        server_start_time = datetime.now()
    
    uptime = datetime.now() - server_start_time
    return str(uptime)

Logging and metrics

Proper logging is essential for detecting and resolving problems early, so make sure it is in place

import structlog
from prometheus_client import Counter, Histogram, generate_latest

# Structured logging configuration
logger = structlog.get_logger()

# Prometheus metrics
tool_calls_total = Counter(
    'mcp_tool_calls_total',
    'Total number of tool calls',
    ['tool_name']
)
tool_duration_seconds = Histogram(
    'mcp_tool_duration_seconds',
    'Tool execution duration',
    ['tool_name']
)

@mcp.tool()
async def monitored_tool(param: str) -> str:
    """Tool with monitoring built in"""
    start_time = time.time()
    
    try:
        # Increment the tool-call counter
        tool_calls_total.labels(tool_name='monitored_tool').inc()
        
        # The actual work
        result = await process_something(param)
        
        # Log success
        logger.info(
            "tool_executed",
            tool_name="monitored_tool",
            param=param,
            duration=time.time() - start_time
        )
        
        return result
        
    except Exception as e:
        # Log the error
        logger.error(
            "tool_failed",
            tool_name="monitored_tool",
            param=param,
            error=str(e),
            duration=time.time() - start_time
        )
        raise
    
    finally:
        # Record the execution time
        tool_duration_seconds.labels(
            tool_name='monitored_tool'
        ).observe(time.time() - start_time)

Chapter 7: Outlook and Conclusion

The future of MCP

We believe MCP will continue to play a central role in the AI agent ecosystem. Even if some other specification were to appear, something like MCP will remain essential technology.

We have raced through the journey from the stdio era to Streamable HTTP; from here, we expect development along the following lines.

First, tool interoperability will improve further. Sharing and combining tools across AI models from different vendors will become commonplace. Security features should also mature: authentication, authorization, audit logging, and other enterprise capabilities are likely to be standardized.

We can also expect better real-time characteristics. Support for more efficient bidirectional protocols such as WebSockets and WebTransport may be added.

Recommendations for MCP implementers today

For developers implementing MCP servers as of 2025 — including our own engineers — we recommend the following approach

First, make active use of (semi-)official SDKs like FastMCP.

I actually started with a hand-rolled implementation myself and, sure enough, ran into interoperability problems and regretted it...

You can focus on business logic without being whipsawed by spec details. This really matters

Second, adopt the Streamable HTTP transport as the standard for remote MCP servers.

It is the most mature and practical choice.

Third, build with production in mind from the very start. Error handling, logging, and monitoring should be designed in from the beginning, not bolted on later. And if you expose the server on the internet, you will also need authentication and authorization.

Conclusion

Today we walked through how the MCP specification has evolved and how to implement it.

MCP has established itself as an important standard for connecting AI with external systems. The journey from the early stdio-based implementation, through HTTP+SSE, to today's Streamable HTTP was a natural evolution shaped by feedback from real-world use.

We also covered practical ground: server implementation with FastMCP, integration with each LLM provider, and operational considerations.

We touched as well on the significance of MCP servers streaming their responses. This is not just a matter of communication efficiency — it holds the potential for a new execution model, and a new UX, in which the AI model, the MCP server, and the application share the same timeline and work together through incremental exchanges. If we had to compare, it feels like the excitement when Ajax arrived in the early 2000s.

So now that the spec has largely stabilized, isn't this the perfect moment to start building innovative AI applications on MCP?

Thank you for reading!

See you next time!