{"id":270,"date":"2026-01-15T05:31:24","date_gmt":"2026-01-14T21:31:24","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=270"},"modified":"2026-01-15T05:31:24","modified_gmt":"2026-01-14T21:31:24","slug":"how-to-build-a-stateless-secure-and-asynchronous-mcp-style-protocol-for-scalable-agent-workflows","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=270","title":{"rendered":"How to Build a Stateless, Secure, and Asynchronous MCP-Style Protocol for Scalable Agent Workflows"},"content":{"rendered":"

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. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
import asyncio, time, json, uuid, hmac, hashlib\nfrom dataclasses import dataclass\nfrom typing import Any, Dict, Optional, Literal, List\nfrom pydantic import BaseModel, Field, ValidationError, ConfigDict\n\n\ndef _now_ms():\n   return int(time.time() * 1000)\n\n\ndef _uuid():\n   return str(uuid.uuid4())\n\n\ndef _canonical_json(obj):\n   return json.dumps(obj, separators=(\",\", \":\"), sort_keys=True).encode()\n\n\ndef _hmac_hex(secret, payload):\n   return hmac.new(secret, _canonical_json(payload), hashlib.sha256).hexdigest()<\/code><\/pre>\n<\/div>\n<\/div>\n
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. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class MCPEnvelope(BaseModel):\n   model_config = ConfigDict(extra=\"forbid\")\n   v: Literal[\"mcp\/0.1\"] = \"mcp\/0.1\"\n   request_id: str = Field(default_factory=_uuid)\n   ts_ms: int = Field(default_factory=_now_ms)\n   client_id: str\n   server_id: str\n   tool: str\n   args: Dict[str, Any] = Field(default_factory=dict)\n   nonce: str = Field(default_factory=_uuid)\n   signature: str\n\n\nclass MCPResponse(BaseModel):\n   model_config = ConfigDict(extra=\"forbid\")\n   v: Literal[\"mcp\/0.1\"] = \"mcp\/0.1\"\n   request_id: str\n   ts_ms: int = Field(default_factory=_now_ms)\n   ok: bool\n   server_id: str\n   status: Literal[\"ok\", \"accepted\", \"running\", \"done\", \"error\"]\n   result: Optional[Dict[str, Any]] = None\n   error: Optional[str] = None\n   signature: str<\/code><\/pre>\n<\/div>\n<\/div>\n
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. It ensures consistent contracts between clients and servers, which is critical for SDK standardization. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class ServerIdentityOut(BaseModel):\n   model_config = ConfigDict(extra=\"forbid\")\n   server_id: str\n   fingerprint: str\n   capabilities: Dict[str, Any]\n\n\nclass BatchSumIn(BaseModel):\n   model_config = ConfigDict(extra=\"forbid\")\n   numbers: List[float] = Field(min_length=1)\n\n\nclass BatchSumOut(BaseModel):\n   model_config = ConfigDict(extra=\"forbid\")\n   count: int\n   total: float\n\n\nclass StartLongTaskIn(BaseModel):\n   model_config = ConfigDict(extra=\"forbid\")\n   seconds: int = Field(ge=1, le=20)\n   payload: Dict[str, Any] = Field(default_factory=dict)\n\n\nclass PollJobIn(BaseModel):\n   model_config = ConfigDict(extra=\"forbid\")\n   job_id: str<\/code><\/pre>\n<\/div>\n<\/div>\n
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. It makes tool behavior predictable and safe, even when invoked by LLM-driven agents. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
@dataclass\nclass JobState:\n   job_id: str\n   status: str\n   result: Optional[Dict[str, Any]] = None\n   error: Optional[str] = None\n\n\nclass MCPServer:\n   def __init__(self, server_id, secret):\n       self.server_id = server_id\n       self.secret = secret\n       self.jobs = {}\n       self.tasks = {}\n\n\n   def _fingerprint(self):\n       return hashlib.sha256(self.secret).hexdigest()[:16]\n\n\n   async def handle(self, env_dict, client_secret):\n       env = MCPEnvelope(**env_dict)\n       payload = env.model_dump()\n       sig = payload.pop(\"signature\")\n       if _hmac_hex(client_secret, payload) != sig:\n           return {\"error\": \"bad signature\"}\n\n\n       if env.tool == \"server_identity\":\n           out = ServerIdentityOut(\n               server_id=self.server_id,\n               fingerprint=self._fingerprint(),\n               capabilities={\"async\": True, \"stateless\": True},\n           )\n           resp = MCPResponse(\n               request_id=env.request_id,\n               ok=True,\n               server_id=self.server_id,\n               status=\"ok\",\n               result=out.model_dump(),\n               signature=\"\",\n           )\n\n\n       elif env.tool == \"batch_sum\":\n           args = BatchSumIn(**env.args)\n           out = BatchSumOut(count=len(args.numbers), total=sum(args.numbers))\n           resp = MCPResponse(\n               request_id=env.request_id,\n               ok=True,\n               server_id=self.server_id,\n               status=\"ok\",\n               result=out.model_dump(),\n               signature=\"\",\n           )\n\n\n       elif env.tool == \"start_long_task\":\n           args = StartLongTaskIn(**env.args)\n           jid = _uuid()\n           self.jobs[jid] = JobState(jid, \"running\")\n\n\n           async def run():\n               await asyncio.sleep(args.seconds)\n               self.jobs[jid].status = \"done\"\n               self.jobs[jid].result = args.payload\n\n\n           self.tasks[jid] = asyncio.create_task(run())\n           resp = MCPResponse(\n               request_id=env.request_id,\n               ok=True,\n               server_id=self.server_id,\n               status=\"accepted\",\n               result={\"job_id\": jid},\n               signature=\"\",\n           )\n\n\n       elif env.tool == \"poll_job\":\n           args = PollJobIn(**env.args)\n           job = self.jobs[args.job_id]\n           resp = MCPResponse(\n               request_id=env.request_id,\n               ok=True,\n               server_id=self.server_id,\n               status=job.status,\n               result=job.result,\n               signature=\"\",\n           )\n\n\n       payload = resp.model_dump()\n       resp.signature = _hmac_hex(self.secret, payload)\n       return resp.model_dump()<\/code><\/pre>\n<\/div>\n<\/div>\n
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. By returning job identifiers and allowing polling, we demonstrate non-blocking, scalable task execution. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class MCPClient:\n   def __init__(self, client_id, secret, server):\n       self.client_id = client_id\n       self.secret = secret\n       self.server = server\n\n\n   async def call(self, tool, args=None):\n       env = MCPEnvelope(\n           client_id=self.client_id,\n           server_id=self.server.server_id,\n           tool=tool,\n           args=args or {},\n           signature=\"\",\n       ).model_dump()\n       env[\"signature\"] = _hmac_hex(self.secret, {k: v for k, v in env.items() if k != \"signature\"})\n       return await self.server.handle(env, self.secret)\n\n\nasync def demo():\n   server_secret = b\"server_secret\"\n   client_secret = b\"client_secret\"\n   server = MCPServer(\"mcp-server-001\", server_secret)\n   client = MCPClient(\"client-001\", client_secret, server)\n\n\n   print(await client.call(\"server_identity\"))\n   print(await client.call(\"batch_sum\", {\"numbers\": [1, 2, 3]}))\n\n\n   start = await client.call(\"start_long_task\", {\"seconds\": 2, \"payload\": {\"task\": \"demo\"}})\n   jid = start[\"result\"][\"job_id\"]\n\n\n   while True:\n       poll = await client.call(\"poll_job\", {\"job_id\": jid})\n       if poll[\"status\"] == \"done\":\n           print(poll)\n           break\n       await asyncio.sleep(0.5)\n\n\nawait demo()<\/code><\/pre>\n<\/div>\n<\/div>\n
We build a lightweight stateless client that signs each request and interacts with the server through structured envelopes. We demonstrate synchronous calls, input validation failures, and asynchronous task polling in a single flow. It shows how clients can reliably consume MCP-style services in real agent pipelines.<\/p>\n
In conclusion, we showed how MCP evolves from a simple tool-calling interface into a robust protocol suitable for real-world systems. We started tasks asynchronously and poll for results without blocking execution, enforce clear contracts through schema validation, and rely on stateless, signed messages to preserve security and flexibility. Together, these patterns demonstrate how modern MCP-style systems support reliable, enterprise-ready agent workflows while remaining simple, transparent, and easy to extend.<\/p>\n
\n
Check out the\u00a0FULL CODES here<\/a><\/strong>.\u00a0Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0100k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
The post How to Build a Stateless, Secure, and Asynchronous MCP-Style Protocol for Scalable Agent Workflows<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we build a c…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-270","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/270","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=270"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/270\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=270"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=270"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=270"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}