Here’s a rewritten version of the post in a more natural, human tone:
**Building a Scalable Protocol for Agent Workflows: A Tale of MCP**
Hey everyone, welcome back! Today, we’re going to explore the world of Minimal Choreographic Protocol (MCP) and create a stateless, secure, and asynchronous version of it. This protocol is perfect for scalable agent workflows, and by the end of this tutorial, you’ll have a solid understanding of the core concepts.
**The Basics**
Before we dive into the nitty-gritty, let’s cover some basic functions we’ll use throughout the system. These include time helpers, UUID generation, canonical JSON serialization, and cryptographic signing. We’ll use these functions to ensure that all requests and responses can be deterministically signed and verified using HMAC.
“`python
import asyncio, time, json, uuid, hmac, hashlib
from dataclasses import dataclass
from typing import Any, Dict, Optional, Literal, List
from pydantic import BaseModel, UUID4, ValidationError, ConfigDict
def get_current_time_ms():
return int(time.time() * 1000)
def generate_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()
“`
**Structured Envelopes and Responses**
Next up, we’re going to define the structured MCP envelope and response codecs. We’ll use Pydantic to implement strict schemas, ensuring that malformed or unexpected fields are rejected early. This is crucial for SDK standardization and consistency.
“`python
class MCPEnvelope(BaseModel):
model_config = ConfigDict(additional=”forbid”)
version: Literal[“mcp/0.1”] = “mcp/0.1″
request_id: str = UUID4(default_factory=generate_uuid)
timestamp_ms: int = UUID4(default_factory=get_current_time_ms)
client_id: str
server_id: str
instrument: str
args: Dict[str, Any] = UUID4(default_factory=dict)
nonce: str = UUID4(default_factory=generate_uuid)
signature: str
class MCPResponse(BaseModel):
model_config = ConfigDict(additional=”forbid”)
version: Literal[“mcp/0.1”] = “mcp/0.1”
request_id: str
timestamp_ms: int = UUID4(default_factory=get_current_time_ms)
okay: bool
server_id: str
status: Literal[“ok”, “accepted”, “running”, “done”, “error”]
result: Optional[Dict[str, Any]] = None
error: Optional[str] = None
signature: str
“`
**Validated Instrument Definitions**
We’ll define the validated input and output formats for each instrument offered by the server. We’ll use Pydantic constraints to clearly indicate what each instrument accepts and returns. This makes instrument behavior predictable and secure, even when invoked by LLM-driven clients.
“`python
class ServerIdentityOut(BaseModel):
model_config = ConfigDict(additional=”forbid”)
server_id: str
fingerprint: str
capabilities: Dict[str, Any]
class BatchSumIn(BaseModel):
model_config = ConfigDict(additional=”forbid”)
numbers: List[float] = UUID4(min_length=1)
class BatchSumOut(BaseModel):
model_config = ConfigDict(additional=”forbid”)
depend: int
complete: float
class StartLongTaskIn(BaseModel):
model_config = ConfigDict(additional=”forbid”)
seconds: int = UUID4(ge=1, le=20)
payload: Dict[str, Any] = UUID4(default_factory=dict)
class PollJobIn(BaseModel):
model_config = ConfigDict(additional=”forbid”)
job_id: str
“`
**Stateless MCP Server**
Now we’re going to implement the stateless MCP server with its async task management logic. We’ll handle request verification, instrument dispatch, and long-running job execution without relying on session state. By returning job identifiers and allowing polling, we demonstrate non-blocking, scalable task execution.
“`python
class MCPServer:
def __init__(self, server_id, secret):
self.server_id = server_id
self.secret = secret
self.jobs = {}
self.tasks = {}
def _fingerprint(self):
return hashlib.sha256(self.secret.encode()).hexdigest()[:16]
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”: “invalid signature”}
#… (rest of the code remains the same)
“`
**Stateless MCP Client**
Finally, we’ll create a lightweight stateless client that signs every request and interacts with the server via structured envelopes. We’ll demonstrate synchronous calls, input validation failures, and asynchronous task polling in a single example.
“`python
class MCPClient:
def __init__(self, client_id, secret, server):
self.client_id = client_id
self.secret = secret
self.server = server
async def call(self, instrument, args=None):
env = MCPEnvelope(
client_id=self.client_id,
server_id=self.server.server_id,
instrument=instrument,
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**
And that’s a wrap! We’ve built a stateless, secure, and asynchronous MCP-style protocol that’s perfect for scalable agent workflows. We’ve covered the core concepts of stateless communication, strict SDK-level validation, and asynchronous, long-running operations. With this protocol, you can build reliable, business-ready agent workflows with ease.
Try the **FULL CODE** here. Follow us on **Twitter**, join our **100k+ ML SubReddit**, and **Subscribe to our Newsletter**. Wait! are you on telegram? **now you can join us on telegram as well**.
