Building a Stateless, Secure MCP Protocol for Agents

Core Principles of MCP Design

In this tutorial, we build a clean, advanced demonstration of modern MCP design by focusing on three core ideas: stateless communication, strict SDK-level validation, and asynchronous, long-running operations. We implement a minimal MCP-like protocol using structured envelopes, signed requests, and Pydantic-validated tools to show how agents and services can interact safely without relying on persistent sessions.

Setting Up Core Utilities

We set up the core utilities required across the entire system, including time helpers, UUID generation, canonical JSON serialization, and cryptographic signing. We ensure that all requests and responses can be deterministically signed and verified using HMAC.

import asyncio, time, json, uuid, hmac, hashlib
from dataclasses import dataclass
from typing import Any, Dict, Optional, Literal, List
from pydantic import BaseModel, Field, ValidationError, ConfigDict
 
def _now_ms():
   return int(time.time() * 1000)
 
def _uuid():
   return str(uuid.uuid4())
 
def _canonical_json(obj):
   return json.dumps(obj, separators=(',', ':'), sort_keys=True).encode()
 
def _hmac_hex(secret, payload):
   return hmac.new(secret, _canonical_json(payload), hashlib.sha256).hexdigest()

Defining MCP Envelope and Response

We define the structured MCP envelope and response formats that every interaction follows. We enforce strict schemas using Pydantic to guarantee that malformed or unexpected fields are rejected early.

class MCPEnvelope(BaseModel):
   model_config = ConfigDict(extra='forbid')
   v: Literal['mcp/0.1'] = 'mcp/0.1'
   request_id: str = Field(default_factory=_uuid)
   ts_ms: int = Field(default_factory=_now_ms)
   client_id: str
   server_id: str
   tool: str
   args: Dict[str, Any] = Field(default_factory=dict)
   nonce: str = Field(default_factory=_uuid)
   signature: str
 
class MCPResponse(BaseModel):
   model_config = ConfigDict(extra='forbid')
   v: Literal['mcp/0.1'] = 'mcp/0.1'
   request_id: str
   ts_ms: int = Field(default_factory=_now_ms)
   ok: bool
   server_id: str
   status: Literal['ok', 'accepted', 'running', 'done', 'error']
   result: Optional[Dict[str, Any]] = None
   error: Optional[str] = None
   signature: str

Validating Input and Output Models

We declare the validated input and output models for each tool exposed by the server. We use Pydantic constraints to clearly express what each tool accepts and returns.

class ServerIdentityOut(BaseModel):
   model_config = ConfigDict(extra='forbid')
   server_id: str
   fingerprint: str
   capabilities: Dict[str, Any]
 
class BatchSumIn(BaseModel):
   model_config = ConfigDict(extra='forbid')
   numbers: List[float] = Field(min_length=1)
 
class BatchSumOut(BaseModel):
   model_config = ConfigDict(extra='forbid')
   count: int
   total: float
 
class StartLongTaskIn(BaseModel):
   model_config = ConfigDict(extra='forbid')
   seconds: int = Field(ge=1, le=20)
   payload: Dict[str, Any] = Field(default_factory=dict)
 
class PollJobIn(BaseModel):
   model_config = ConfigDict(extra='forbid')
   job_id: str

Stateless MCP Server Implementation

We implement the stateless MCP server along with its async task management logic. We handle request verification, tool dispatch, and long-running job execution without relying on session state.

class MCPServer:
   def __init__(self, server_id, secret):
       self.server_id = server_id
       self.secret = secret
       self.jobs = {}
       self.tasks = {}
 
   async def handle(self, env_dict, client_secret):
       env = MCPEnvelope(**env_dict)
       payload = env.model_dump()
       sig = payload.pop('signature')
       if _hmac_hex(client_secret, payload) != sig:
           return {'error': 'bad signature'}
 
       
       # Handle different tools

Building the Stateless Client

We build a lightweight stateless client that signs each request and interacts with the server through structured envelopes.

class MCPClient:
   def __init__(self, client_id, secret, server):
       self.client_id = client_id
       self.secret = secret
       self.server = server
 
   async def call(self, tool, args=None):
       env = MCPEnvelope(
           client_id=self.client_id,
           server_id=self.server.server_id,
           tool=tool,
           args=args or {},
           signature='',
       ).model_dump()
       env['signature'] = _hmac_hex(self.secret, {k: v for k, v in env.items() if k != 'signature'})
       return await self.server.handle(env, self.secret)

Conclusion

We showed how MCP evolves from a simple tool-calling interface into a robust protocol suitable for real-world systems. Together, these patterns demonstrate how modern MCP-style systems support reliable, enterprise-ready agent workflows while remaining simple, transparent, and easy to extend.