Add files via upload

Added tutorial on "How to Build a Stateless, Secure, and Asynchronous MCP-Style Protocol for Scalable Agent Workflows"

Author: teammtp

MCP Codes/stateless_async_mcp_protocol_demo_Marktechpost.ipynb

@@ -0,0 +1,287 @@
+{
+ "nbformat": 4,
+ "nbformat_minor": 0,
+ "metadata": {
+ "colab": {
+ "provenance": []
+ },
+ "kernelspec": {
+ "name": "python3",
+ "display_name": "Python 3"
+ },
+ "language_info": {
+ "name": "python"
+ }
+ },
+ "cells": [
+ {
+ "cell_type": "code",
+ "source": [
+ "import asyncio, time, json, uuid, hmac, hashlib\n",
+ "from dataclasses import dataclass\n",
+ "from typing import Any, Dict, Optional, Literal, List\n",
+ "from pydantic import BaseModel, Field, ValidationError, ConfigDict\n",
+ "\n",
+ "def _now_ms():\n",
+ " return int(time.time() * 1000)\n",
+ "\n",
+ "def _uuid():\n",
+ " return str(uuid.uuid4())\n",
+ "\n",
+ "def _canonical_json(obj):\n",
+ " return json.dumps(obj, separators=(\",\", \":\"), sort_keys=True).encode()\n",
+ "\n",
+ "def _hmac_hex(secret, payload):\n",
+ " return hmac.new(secret, _canonical_json(payload), hashlib.sha256).hexdigest()"
+ ],
+ "metadata": {
+ "id": "jd_g-C6kqngD"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "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",
+ "class 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"
+ ],
+ "metadata": {
+ "id": "S_TwMtR5qnWI"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "class ServerIdentityOut(BaseModel):\n",
+ " model_config = ConfigDict(extra=\"forbid\")\n",
+ " server_id: str\n",
+ " fingerprint: str\n",
+ " capabilities: Dict[str, Any]\n",
+ "\n",
+ "class BatchSumIn(BaseModel):\n",
+ " model_config = ConfigDict(extra=\"forbid\")\n",
+ " numbers: List[float] = Field(min_length=1)\n",
+ "\n",
+ "class BatchSumOut(BaseModel):\n",
+ " model_config = ConfigDict(extra=\"forbid\")\n",
+ " count: int\n",
+ " total: float\n",
+ "\n",
+ "class 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",
+ "class PollJobIn(BaseModel):\n",
+ " model_config = ConfigDict(extra=\"forbid\")\n",
+ " job_id: str"
+ ],
+ "metadata": {
+ "id": "fd9g8VQyqnTY"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "@dataclass\n",
+ "class JobState:\n",
+ " job_id: str\n",
+ " status: str\n",
+ " result: Optional[Dict[str, Any]] = None\n",
+ " error: Optional[str] = None\n",
+ "\n",
+ "class 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",
+ " def _fingerprint(self):\n",
+ " return hashlib.sha256(self.secret).hexdigest()[:16]\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",
+ " 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",
+ " 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",
+ " elif env.tool == \"start_long_task\":\n",
+ " args = StartLongTaskIn(**env.args)\n",
+ " jid = _uuid()\n",
+ " self.jobs[jid] = JobState(jid, \"running\")\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",
+ " 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",
+ " 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",
+ " payload = resp.model_dump()\n",
+ " resp.signature = _hmac_hex(self.secret, payload)\n",
+ " return resp.model_dump()"
+ ],
+ "metadata": {
+ "id": "Fap0YgEiqnQs"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 11,
+ "metadata": {
+ "id": "VE1XPz73I8Zt",
+ "colab": {
+ "base_uri": "https://localhost:8080/"
+ },
+ "outputId": "b700fa90-bf4c-49d4-bec7-f198acb17151"
+ },
+ "outputs": [
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "=== 1) SERVER IDENTITY (stateless envelope + signed response) ===\n",
+ "raw: {'v': 'mcp/0.1', 'request_id': '552f2dd3-a2d2-4814-9c9a-de0d18b3a09a', 'ts_ms': 1766383443221, 'ok': True, 'server_id': 'mcp-server-prod-001', 'status': 'ok', 'result': {'server_id': 'mcp-server-prod-001', 'fingerprint': '937b0da37c69a0d4', 'capabilities': {'async_jobs': True, 'stateless_envelopes': True, 'sdk_validation': True, 'tools': ['server_identity', 'start_long_task', 'poll_job', 'batch_sum']}}, 'error': None, 'signature': '34bc71bb1ebac2612e110720338631b896aebdd765f5235dc5b243c4cc1aeb6c'}\n",
+ "server signature ok?: True\n",
+ "\n",
+ "=== 2) SDK VALIDATION (good batch_sum) ===\n",
+ "raw: {'v': 'mcp/0.1', 'request_id': 'a983e5ed-028b-478e-9ed8-d0423d9abaac', 'ts_ms': 1766383443222, 'ok': True, 'server_id': 'mcp-server-prod-001', 'status': 'ok', 'result': {'count': 4, 'total': 16.5}, 'error': None, 'signature': 'ea1c8c96302a396557cbe69e4c9b274abe361659c8e9d6ee7ce0c099254436e4'}\n",
+ "server signature ok?: True\n",
+ "\n",
+ "=== 3) SDK VALIDATION (bad batch_sum -> error) ===\n",
+ "raw: {'v': 'mcp/0.1', 'request_id': 'd362ffcc-d153-4d6f-89f9-eb954203c8cd', 'ts_ms': 1766383443222, 'ok': False, 'server_id': 'mcp-server-prod-001', 'status': 'error', 'result': None, 'error': \"Tool args validation error: [{'type': 'too_short', 'loc': ('numbers',), 'msg': 'List should have at least 1 item after validation, not 0', 'input': [], 'ctx': {'field_type': 'List', 'min_length': 1, 'actual_length': 0}, 'url': 'https://errors.pydantic.dev/2.12/v/too_short'}]\", 'signature': '34591661fc91d836d1a3fc143937248b65eb2e0d22e9659a18e76d40855a9799'}\n",
+ "server signature ok?: True\n",
+ "\n",
+ "=== 4) ASYNC LONG-RUNNING OP (start -> poll -> done) ===\n",
+ "start: {'v': 'mcp/0.1', 'request_id': '90f182e6-cd3e-4cef-acf7-999780255cfb', 'ts_ms': 1766383443222, 'ok': True, 'server_id': 'mcp-server-prod-001', 'status': 'accepted', 'result': {'job_id': '6b40192d-b686-46ea-8bc5-193e5087d20a', 'status': 'running'}, 'error': None, 'signature': '34da5c27cec1a720ec024bb81299f15e5ec6876a94898d14879da38f668ef781'}\n",
+ "poll: running\n",
+ "poll: running\n",
+ "poll: running\n",
+ "poll: running\n",
+ "poll: running\n",
+ "poll: done\n",
+ "final: {'v': 'mcp/0.1', 'request_id': '0b6c840a-b658-431e-9939-c3badd686af0', 'ts_ms': 1766383446734, 'ok': True, 'server_id': 'mcp-server-prod-001', 'status': 'done', 'result': {'job_id': '6b40192d-b686-46ea-8bc5-193e5087d20a', 'status': 'done', 'result': {'echo_payload': {'job': 'recompute_embeddings', 'scope': 'Q4'}, 'duration_s': 3, 'completed_ts_ms': 1766383446223}, 'error': None}, 'error': None, 'signature': 'ff92745c64f07c002cc4fb07fd71bbae55f338aafb5274c763f44ddd93bb4f5f'}\n",
+ "\n",
+ "✅ Demo complete: async ops + stateless envelopes + strict SDK validation.\n"
+ ]
+ }
+ ],
+ "source": [
+ "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",
+ " 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",
+ "async 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",
+ " print(await client.call(\"server_identity\"))\n",
+ " print(await client.call(\"batch_sum\", {\"numbers\": [1, 2, 3]}))\n",
+ "\n",
+ " start = await client.call(\"start_long_task\", {\"seconds\": 2, \"payload\": {\"task\": \"demo\"}})\n",
+ " jid = start[\"result\"][\"job_id\"]\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",
+ "await demo()"
+ ]
+ }
+ ]
+}