{"id":601,"date":"2026-03-24T02:33:06","date_gmt":"2026-03-23T18:33:06","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=601"},"modified":"2026-03-24T02:33:06","modified_gmt":"2026-03-23T18:33:06","slug":"how-to-design-a-production-ready-ai-agent-that-automates-google-colab-workflows-using-colab-mcp-mcp-tools-fastmcp-and-kernel-execution","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=601","title":{"rendered":"How to Design a Production-Ready AI Agent That Automates Google Colab Workflows Using Colab-MCP, MCP Tools, FastMCP, and Kernel Execution"},"content":{"rendered":"

In this tutorial, we build an advanced, hands-on tutorial around Google\u2019s newly released colab-mcp<\/strong><\/a>, an open-source MCP (Model Context Protocol) server that lets any AI agent programmatically control Google Colab notebooks and runtimes. Across five self-contained snippets, we go from first principles to production-ready patterns. We start by constructing a minimal MCP tool registry from scratch. Hence, we understand the protocol\u2019s core mechanics, tool registration, schema generation, and async dispatch, before graduating to the real FastMCP framework that colab-mcp is built on. We then simulate both of the server\u2019s operational modes: the Session Proxy mode, where we spin up an authenticated WebSocket bridge between a browser frontend and an MCP client, and the Runtime mode, where we wire up a direct kernel execution engine with persistent state, lazy initialization, and Jupyter-style output handling. From there, we assemble a complete AI agent loop that reasons about tasks, selects tools, executes code, inspects results, and iterates, the same pattern Claude Code and Gemini CLI use when connected to colab-mcp in the real world. We close with production-grade orchestration: automatic retries with exponential backoff, timeout handling, dependency-aware cell sequencing, and execution reporting.<\/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 subprocess, sys\n\n\ndef install(pkg):\n   subprocess.check_call([sys.executable, \"-m\", \"pip\", \"install\", \"-q\", pkg])\n\n\ninstall(\"fastmcp>=2.2.0,<3.0.0\")\ninstall(\"websockets>=15.0.1\")\ninstall(\"pydantic>=2.0.0,<3.0.0\")\ninstall(\"requests>=2.32.0\")\ninstall(\"mcp>=1.0.0\")\ninstall(\"httpx\")\ninstall(\"google-auth\")\ninstall(\"google-auth-oauthlib\")\ninstall(\"openai\")\n\n\nprint(\"\"\u2705\" All dependencies installed.\")\n\n\nARCHITECTURE_OVERVIEW = \"\"\"\n\u2554\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2557\n\u2551                    colab-mcp Architecture                          \u2551\n\u2560\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2563\n\u2551                                                                    \u2551\n\u2551  \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510     MCP (JSON-RPC)     \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510     \u2551\n\u2551  \u2502  AI Agent    \u2502\u25c4\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u25ba\u2502  colab-mcp       \u2502     \u2551\n\u2551  \u2502  (Claude,    \u2502    stdio transport      \u2502  FastMCP Server  \u2502     \u2551\n\u2551  \u2502   Gemini,    \u2502                         \u2502                  \u2502     \u2551\n\u2551  \u2502   Custom)    \u2502                         \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518     \u2551\n\u2551  \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518                                \u2502                 \u2551\n\u2551                                    \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u253c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510    \u2551\n\u2551                                    \u2502             \u2502            \u2502    \u2551\n\u2551                              \u250c\u2500\u2500\u2500\u2500\u2500\u25bc\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u25bc\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u2502    \u2551\n\u2551                              \u2502  SESSION   \u2502 \u2502   RUNTIME    \u2502 \u2502    \u2551\n\u2551                              \u2502  PROXY     \u2502 \u2502   MODE       \u2502 \u2502    \u2551\n\u2551                              \u2502  MODE      \u2502 \u2502              \u2502 \u2502    \u2551\n\u2551                              \u2502            \u2502 \u2502  Jupyter     \u2502 \u2502    \u2551\n\u2551                              \u2502  WebSocket \u2502 \u2502  Kernel      \u2502 \u2502    \u2551\n\u2551                              \u2502  Bridge    \u2502 \u2502  Client      \u2502 \u2502    \u2551\n\u2551                              \u2514\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2502    \u2551\n\u2551                                    \u2502             \u2502            \u2502    \u2551\n\u2551                              \u250c\u2500\u2500\u2500\u2500\u2500\u25bc\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u25bc\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u2502    \u2551\n\u2551                              \u2502  Browser   \u2502 \u2502  Colab VM    \u2502 \u2502    \u2551\n\u2551                              \u2502  Colab UI  \u2502 \u2502  (GPU\/TPU)   \u2502 \u2502    \u2551\n\u2551                              \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2502    \u2551\n\u2551                                                               \u2502    \u2551\n\u2551  SESSION PROXY (default):  Browser \"\u2194\" WebSocket \"\u2194\" Agent       \u2502    \u2551\n\u2551  RUNTIME MODE (opt-in):    Agent \u2192 Kernel \u2192 Code Execution   \u2502    \u2551\n\u255a\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u255d\n\"\"\"\nprint(ARCHITECTURE_OVERVIEW)\n\n\nimport asyncio\nimport json\nfrom typing import Any\n\n\nclass MCPToolRegistry:\n\n\n   def __init__(self, name: str):\n       self.name = name\n       self._tools: dict[str, dict] = {}\n\n\n   def tool(self, func):\n       import inspect\n       sig = inspect.signature(func)\n       params = {}\n       for pname, p in sig.parameters.items():\n           ptype = \"string\"\n           if p.annotation == int:\n               ptype = \"integer\"\n           elif p.annotation == bool:\n               ptype = \"boolean\"\n           elif p.annotation == float:\n               ptype = \"number\"\n           params[pname] = {\"type\": ptype, \"description\": f\"Parameter: {pname}\"}\n\n\n       self._tools[func.__name__] = {\n           \"name\": func.__name__,\n           \"description\": func.__doc__ or \"\",\n           \"inputSchema\": {\n               \"type\": \"object\",\n               \"properties\": params,\n               \"required\": list(params.keys())\n           },\n           \"handler\": func,\n       }\n       return func\n\n\n   def list_tools(self) -> list[dict]:\n       return [\n           {k: v for k, v in t.items() if k != \"handler\"}\n           for t in self._tools.values()\n       ]\n\n\n   async def call_tool(self, name: str, arguments: dict) -> Any:\n       if name not in self._tools:\n           raise ValueError(f\"Unknown tool: {name}\")\n       handler = self._tools[name][\"handler\"]\n       if asyncio.iscoroutinefunction(handler):\n           return await handler(**arguments)\n       return handler(**arguments)\n\n\n\n\nserver = MCPToolRegistry(\"colab-mcp-demo\")\n\n\n@server.tool\ndef execute_code(code: str) -> str:\n   \"\"\"Execute Python code in the runtime kernel and return output.\"\"\"\n   import io, contextlib\n   buf = io.StringIO()\n   try:\n       with contextlib.redirect_stdout(buf):\n           exec(code, {\"__builtins__\": __builtins__})\n       output = buf.getvalue()\n       return output if output else \"(no output)\"\n   except Exception as e:\n       return f\"Error: {type(e).__name__}: {e}\"\n\n\n@server.tool\ndef add_code_cell(code: str, cell_index: int) -> str:\n   \"\"\"Add a code cell to the notebook at the specified index.\"\"\"\n   return json.dumps({\n       \"status\": \"success\",\n       \"action\": \"add_code_cell\",\n       \"cell_index\": cell_index,\n       \"preview\": code[:80] + (\"...\" if len(code) > 80 else \"\"),\n   })\n\n\n@server.tool\ndef add_text_cell(content: str, cell_index: int) -> str:\n   \"\"\"Add a markdown cell to the notebook at the specified index.\"\"\"\n   return json.dumps({\n       \"status\": \"success\",\n       \"action\": \"add_text_cell\",\n       \"cell_index\": cell_index,\n       \"preview\": content[:80] + (\"...\" if len(content) > 80 else \"\"),\n   })\n\n\n@server.tool\ndef get_cells(cell_index_start: int, include_outputs: bool) -> str:\n   \"\"\"Retrieve cells from the notebook starting at the given index.\"\"\"\n   return json.dumps({\n       \"cells\": [\n           {\"cell_type\": \"code\", \"id\": \"cell_0\", \"source\": [\"import pandas as pd\"]},\n           {\"cell_type\": \"markdown\", \"id\": \"cell_1\", \"source\": [\"# Analysis\"]},\n       ]\n   })\n\n\nprint(\"\"\ud83d\udccb\" Registered MCP Tools:\")\nprint(\"=\" * 60)\nfor tool in server.list_tools():\n   print(f\"n\"\ud83d\udd27\" {tool['name']}\")\n   print(f\"   Description: {tool['description']}\")\n   params = tool['inputSchema']['properties']\n   for pname, pinfo in params.items():\n       print(f\"   Param: {pname} ({pinfo['type']})\")\n\n\nprint(\"nn\"\ud83d\ude80\" Calling Tools:\")\nprint(\"=\" * 60)\n\n\nasync def demo_tool_calls():\n   result = await server.call_tool(\"execute_code\", {\n       \"code\": \"print('Hello from the MCP runtime!')nprint(2 + 2)\"\n   })\n   print(f\"nexecute_code result:n{result}\")\n\n\n   result = await server.call_tool(\"add_code_cell\", {\n       \"code\": \"import matplotlib.pyplot as pltnplt.plot([1,2,3],[1,4,9])nplt.show()\",\n       \"cell_index\": 0,\n   })\n   print(f\"nadd_code_cell result:n{result}\")\n\n\n   result = await server.call_tool(\"get_cells\", {\n       \"cell_index_start\": 0,\n       \"include_outputs\": False,\n   })\n   print(f\"nget_cells result:n{result}\")\n\n\ntry:\n   import nest_asyncio\n   nest_asyncio.apply()\nexcept ImportError:\n   subprocess.check_call([sys.executable, \"-m\", \"pip\", \"install\", \"-q\", \"nest_asyncio\"])\n   import nest_asyncio\n   nest_asyncio.apply()\n\n\nasyncio.run(demo_tool_calls())<\/code><\/pre>\n<\/div>\n<\/div>\n
We install all the dependencies the tutorial needs: FastMCP, websockets, Pydantic, the MCP SDK, and Google auth libraries, so the remaining snippets run without interruption. We then build a custom MCPToolRegistry class entirely from scratch, walking ourselves through the exact mechanics the protocol relies on: decorator-based tool registration, automatic JSON Schema generation from Python type hints, and async tool dispatch. We register four tools that mirror the real colab-mcp surface, execute_code, add_code_cell, add_text_cell, and get_cells, list their schemas, and call each one to confirm the full request-response cycle works end to end.<\/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
from fastmcp import FastMCP\nimport asyncio\nimport json\nimport secrets\nimport websockets\nfrom websockets.asyncio.server import serve as ws_serve\nimport nest_asyncio\nnest_asyncio.apply()\n\n\nmcp = FastMCP(\"colab-mcp-tutorial\")\n\n\n@mcp.tool()\ndef open_colab_browser_connection() -> dict:\n   \"\"\"Opens a connection to the Colab browser UI.\"\"\"\n   token = secrets.token_hex(16)\n   port = 8765\n   url = f\"https:\/\/colab.research.google.com\/scratchpads#mcpProxyToken={token}&mcpProxyPort={port}\"\n   return {\n       \"result\": True,\n       \"message\": \"Browser connection established\",\n       \"url\": url,\n       \"token\": token,\n       \"port\": port,\n   }\n\n\n@mcp.tool()\ndef proxy_get_cells(cell_index_start: int = 0, include_outputs: bool = True) -> dict:\n   \"\"\"Get notebook cells from the connected Colab frontend.\"\"\"\n   return {\n       \"cells\": [\n           {\n               \"cell_type\": \"code\",\n               \"id\": \"abc123\",\n               \"source\": [\"import numpy as npn\", \"data = np.random.randn(100)n\"],\n               \"outputs\": [{\"output_type\": \"execute_result\", \"text\": \"array([...])\"}]\n                   if include_outputs else [],\n           },\n           {\n               \"cell_type\": \"markdown\",\n               \"id\": \"def456\",\n               \"source\": [\"# Data Analysis Reportn\"],\n               \"outputs\": [],\n           },\n       ]\n   }\n\n\n@mcp.tool()\ndef proxy_add_code_cell(cell_index: int, code: str, language: str = \"python\") -> dict:\n   \"\"\"Add a new code cell to the notebook at the specified position.\"\"\"\n   return {\"status\": \"ok\", \"cell_index\": cell_index, \"language\": language}\n\n\n@mcp.tool()\ndef proxy_add_text_cell(cell_index: int, content: str) -> dict:\n   \"\"\"Add a new markdown cell to the notebook at the specified position.\"\"\"\n   return {\"status\": \"ok\", \"cell_index\": cell_index}\n\n\n@mcp.tool()\ndef proxy_execute_cell(cell_index: int) -> dict:\n   \"\"\"Execute the cell at the specified index in the connected notebook.\"\"\"\n   return {\"status\": \"ok\", \"cell_index\": cell_index, \"execution_count\": 1}\n\n\n@mcp.tool()\ndef runtime_execute_code(code: str) -> dict:\n   \"\"\"Execute Python code directly in a Colab kernel (Runtime Mode).\"\"\"\n   import io, contextlib, traceback\n   stdout_buf = io.StringIO()\n   stderr_buf = io.StringIO()\n\n\n   try:\n       with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf):\n           exec(code, {\"__builtins__\": __builtins__})\n       return {\n           \"outputs\": [\n               {\"output_type\": \"stream\", \"name\": \"stdout\", \"text\": stdout_buf.getvalue()},\n           ]\n       }\n   except Exception:\n       return {\n           \"outputs\": [\n               {\"output_type\": \"error\", \"traceback\": traceback.format_exc()},\n           ]\n       }\n\n\nprint(\"\"\ud83d\udccb\" FastMCP Server Tools:\")\nprint(\"=\" * 60)\n\n\nasync def list_fastmcp_tools():\n   tools_dict = await mcp.get_tools()\n   for name, tool in tools_dict.items():\n       print(f\"n\"\ud83d\udd27\" {tool.name}\")\n       print(f\"   {tool.description[:100]}\")\n   return tools_dict\n\n\ntools = asyncio.run(list_fastmcp_tools())\nprint(f\"n\"\u2705\" Total tools registered: {len(tools)}\")\n\n\n\n\nclass SimulatedColabWebSocketServer:\n\n\n   def __init__(self, host: str = \"localhost\", port: int = 0):\n       self.host = host\n       self.port = port\n       self.token = secrets.token_hex(16)\n       self.connection_live = asyncio.Event()\n       self._server = None\n       self._messages_received: list[dict] = []\n\n\n   async def _handler(self, websocket):\n       try:\n           auth_msg = await asyncio.wait_for(websocket.recv(), timeout=10.0)\n           auth_data = json.loads(auth_msg)\n\n\n           if auth_data.get(\"token\") != self.token:\n               await websocket.send(json.dumps({\"error\": \"Invalid token\"}))\n               await websocket.close()\n               return\n\n\n           await websocket.send(json.dumps({\"status\": \"authenticated\"}))\n           self.connection_live.set()\n           print(f\"   \"\u2705\" Client authenticated (token: {self.token[:8]}...)\")\n\n\n           async for message in websocket:\n               data = json.loads(message)\n               self._messages_received.append(data)\n               print(f\"   \"\ud83d\udce8\" Received: {data.get('method', 'unknown')} \"\n                     f\"\u2014 {json.dumps(data.get('params', {}))[:80]}\")\n\n\n               response = {\n                   \"jsonrpc\": \"2.0\",\n                   \"id\": data.get(\"id\"),\n                   \"result\": {\"status\": \"ok\", \"tool\": data.get(\"method\")},\n               }\n               await websocket.send(json.dumps(response))\n\n\n       except websockets. exceptions.ConnectionClosed:\n           print(\"   \"\u26a1\" Connection closed\")\n       except asyncio.TimeoutError:\n           print(\"   \"\u23f0\" Auth timeout\")\n\n\n   async def start(self):\n       self._server = await ws_serve(self._handler, self.host, self.port)\n       self.port = self._server.sockets[0].getsockname()[1]\n       print(f\"   \"\ud83c\udf10\" WebSocket server running on ws:\/\/{self.host}:{self.port}\")\n       print(f\"   \"\ud83d\udd11\" Token: {self.token[:8]}...\")\n       return self\n\n\n   async def stop(self):\n       if self._server:\n           self._server.close()\n           await self._server.wait_closed()\n           print(\"   \"\ud83d\uded1\" WebSocket server stopped\")\n\n\n\n\nasync def simulate_browser_client(port: int, token: str):\n   uri = f\"ws:\/\/localhost:{port}\"\n   async with websockets.connect(uri) as ws:\n       await ws.send(json.dumps({\"token\": token}))\n       auth_response = json.loads(await ws.recv())\n       print(f\"   \"\ud83d\udd10\" Auth response: {auth_response}\")\n\n\n       tool_call = {\n           \"jsonrpc\": \"2.0\",\n           \"id\": 1,\n           \"method\": \"add_code_cell\",\n           \"params\": {\n               \"cellIndex\": 0,\n               \"code\": \"import pandas as pdndf = pd.read_csv('data.csv')\",\n               \"language\": \"python\",\n           }\n       }\n       await ws.send(json.dumps(tool_call))\n       response = json.loads(await ws.recv())\n       print(f\"   \"\ud83d\udcec\" Tool response: {response}\")\n\n\n       execute_call = {\n           \"jsonrpc\": \"2.0\",\n           \"id\": 2,\n           \"method\": \"execute_cell\",\n           \"params\": {\"cellIndex\": 0}\n       }\n       await ws.send(json.dumps(execute_call))\n       response = json.loads(await ws.recv())\n       print(f\"   \"\ud83d\udcec\" Execute response: {response}\")\n\n\n\n\nasync def run_websocket_demo():\n   print(\"\"\ud83c\udf09\" WebSocket Bridge Demo (Session Proxy Mode)\")\n   print(\"=\" * 60)\n\n\n   print(\"n\"\ud83d\udce1\" Starting WebSocket server...\")\n   wss = SimulatedColabWebSocketServer()\n   await wss.start()\n\n\n   print(\"n\"\ud83d\udda5\"  Simulating browser frontend connection...\")\n   await simulate_browser_client(wss.port, wss.token)\n\n\n   print(f\"n\"\ud83d\udcca\" Server received {len(wss._messages_received)} tool calls\")\n\n\n   await wss.stop()\n   print(\"n\"\u2705\" Demo complete!\")\n\n\nasyncio.run(run_websocket_demo())<\/code><\/pre>\n<\/div>\n<\/div>\n
We graduate from our hand-rolled registry to the real FastMCP framework and create a server with six tools spanning both of colab-mcp\u2019s operational modes: proxy tools like open_colab_browser_connection, proxy_add_code_cell, and proxy_execute_cell, plus runtime_execute_code for direct kernel access. We then build a SimulatedColabWebSocketServer that replicates the Session Proxy architecture \u2014 it listens for connections, validates a security token on the first message, and forwards JSON-RPC tool calls between the browser frontend and the MCP client. We run the full demo by spinning up the server, connecting a simulated browser client that sends add_code_cell and execute_cell calls, and verifying that authenticated messages flow correctly through the bridge.<\/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\nimport io\nimport contextlib\nimport traceback\nimport uuid\nfrom dataclasses import dataclass, field\nfrom typing import Optional\nimport nest_asyncio\nnest_asyncio.apply()\n\n\n@dataclass\nclass KernelOutput:\n   output_type: str\n   text: str = \"\"\n   data: dict = field(default_factory=dict)\n   traceback_lines: list = field(default_factory=list)\n\n\n@dataclass\nclass ExecutionResult:\n   success: bool\n   outputs: list[KernelOutput]\n   execution_count: int\n\n\n\n\nclass ColabRuntimeSimulator:\n\n\n   def __init__(self):\n       self._id = uuid.uuid4()\n       self._execution_count = 0\n       self._namespace: dict = {\"__builtins__\": __builtins__}\n       self._is_started = False\n\n\n   @property\n   def runtime_id(self) -> str:\n       return str(self._id)\n\n\n   async def start(self) -> None:\n       print(f\"   \"\ud83d\udd04\" Initializing runtime {self.runtime_id[:8]}...\")\n       print(f\"   \"\ud83d\udd10\" [Simulated] OAuth2 authentication...\")\n       await asyncio.sleep(0.1)\n       print(f\"   \"\ud83d\udda5\"  [Simulated] Requesting VM assignment...\")\n       await asyncio.sleep(0.1)\n       print(f\"   \"\ud83d\udd0c\" [Simulated] Connecting to Jupyter kernel...\")\n       await asyncio.sleep(0.1)\n       self._is_started = True\n       print(f\"   \"\u2705\" Runtime started!\")\n\n\n   async def execute_code(self, code: str) -> ExecutionResult:\n       if not self._is_started:\n           await self.start()\n\n\n       self._execution_count += 1\n       outputs: list[KernelOutput] = []\n       stdout_buf = io.StringIO()\n       stderr_buf = io.StringIO()\n\n\n       try:\n           with contextlib.redirect_stdout(stdout_buf), \n                contextlib.redirect_stderr(stderr_buf):\n               try:\n                   result = eval(code, self._namespace)\n                   if result is not None:\n                       outputs.append(KernelOutput(\n                           output_type=\"execute_result\",\n                           text=repr(result),\n                           data={\"text\/plain\": repr(result)},\n                       ))\n               except SyntaxError:\n                   exec(code, self._namespace)\n\n\n           stdout_text = stdout_buf.getvalue()\n           if stdout_text:\n               outputs.append(KernelOutput(\n                   output_type=\"stream\",\n                   text=stdout_text,\n               ))\n\n\n           stderr_text = stderr_buf.getvalue()\n           if stderr_text:\n               outputs.append(KernelOutput(\n                   output_type=\"stream\",\n                   text=stderr_text,\n               ))\n\n\n           return ExecutionResult(\n               success=True,\n               outputs=outputs,\n               execution_count=self._execution_count,\n           )\n\n\n       except Exception as e:\n           tb = traceback.format_exc()\n           outputs.append(KernelOutput(\n               output_type=\"error\",\n               text=str(e),\n               traceback_lines=tb.split(\"n\"),\n           ))\n           return ExecutionResult(\n               success=False,\n               outputs=outputs,\n               execution_count=self._execution_count,\n           )\n\n\n   async def stop(self) -> None:\n       if self._is_started:\n           print(f\"   \"\ud83d\uded1\" Unassigning VM for runtime {self.runtime_id[:8]}...\")\n           self._is_started = False\n           print(f\"   \"\u2705\" Runtime stopped and VM released.\")\n\n\n\n\nasync def runtime_demo():\n   print(\"\"\u26a1\" Runtime Mode Demo\")\n   print(\"=\" * 60)\n\n\n   runtime = ColabRuntimeSimulator()\n   await runtime.start()\n\n\n   code_snippets = [\n       \"\"\"\nimport random\n.seed(42)\ndata = [random.gauss(0, 1) for _ in range(1000)]\nprint(f\"Generated {len(data)} data points\")\nprint(f\"Mean: {sum(data)\/len(data):.4f}\")\nprint(f\"Min: {min(data):.4f}, Max: {max(data):.4f}\")\n\"\"\",\n       \"\"\"\nvariance = sum((x - sum(data)\/len(data))**2 for x in data) \/ len(data)\nstd_dev = variance ** 0.5\nprint(f\"Variance: {variance:.4f}\")\nprint(f\"Std Dev: {std_dev:.4f}\")\n\"\"\",\n       \"len(data)\",\n       \"undefined_variable + 1\",\n   ]\n\n\n   for i, code in enumerate(code_snippets):\n       print(f\"n{'\u2500' * 40}\")\n       print(f\"\"\ud83d\udcdd\" Executing cell [{i+1}]:\")\n       print(f\"   Code: {code.strip()[:60]}{'...' if len(code.strip()) > 60 else ''}\")\n\n\n       result = await runtime.execute_code(code)\n\n\n       status = \"\"\u2705\"\" if result.success else \"\"\u274c\"\"\n       print(f\"   {status} Execution #{result.execution_count} \"\n             f\"({'success' if result. success else 'error'})\")\n\n\n       for out in result.outputs:\n           if out.output_type == \"stream\":\n               for line in out.text.strip().split(\"n\"):\n                   print(f\"   \"\ud83d\udce4\" {line}\")\n           elif out.output_type == \"execute_result\":\n               print(f\"   \"\ud83d\udce4\" Out[{result.execution_count}]: {out.text}\")\n           elif out.output_type == \"error\":\n               print(f\"   \"\u26a0\"  {out.text}\")\n\n\n   await runtime.stop()\n   print(f\"n\"\u2705\" Runtime demo complete!\")\n\n\nasyncio.run(runtime_demo())<\/code><\/pre>\n<\/div>\n<\/div>\n
We construct a ColabRuntimeSimulator that mirrors the ColabRuntimeTool from the real codebase, complete with the lazy initialization chain, simulated OAuth2 authentication, VM assignment, and Jupyter kernel connection, that fires only when we first execute code. We run four sequential cells through it, demonstrating that the runtime maintains persistent state across executions: we create a 1,000-point dataset in cell one, compute variance and standard deviation from that same dataset in cell two, evaluate a bare expression in cell three, and trigger an intentional NameError in cell four to confirm errors propagate cleanly. We finish by calling stop(), which simulates VM unassignment and resource cleanup, showing the full lifecycle from kernel startup to graceful shutdown.<\/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\nimport json\nimport io\nimport contextlib\nimport re\nfrom dataclasses import dataclass\nfrom typing import Callable, Awaitable\nimport nest_asyncio\nnest_asyncio.apply()\n\n\nTOOL_DEFINITIONS = [\n   {\n       \"name\": \"execute_code\",\n       \"description\": \"Execute Python code in the Colab kernel. Returns stdout, results, or errors. State persists between calls.\"\n       \"parameters\": {\n           \"type\": \"object\",\n           \"properties\": {\n               \"code\": {\"type\": \"string\", \"description\": \"Python code to execute\"},\n           },\n           \"required\": [\"code\"],\n       }\n   },\n   {\n       \"name\": \"add_code_cell\",\n       \"description\": \"Add a code cell to the notebook at a given index.\",\n       \"parameters\": {\n           \"type\": \"object\",\n           \"properties\": {\n               \"cell_index\": {\"type\": \"integer\", \"description\": \"Position to insert\"},\n               \"code\": {\"type\": \"string\", \"description\": \"Python code for the cell\"},\n           },\n           \"required\": [\"cell_index\", \"code\"],\n       }\n   },\n   {\n       \"name\": \"add_text_cell\",\n       \"description\": \"Add a markdown documentation cell to the notebook.\",\n       \"parameters\": {\n           \"type\": \"object\",\n           \"properties\": {\n               \"cell_index\": {\"type\": \"integer\", \"description\": \"Position to insert\"},\n               \"content\": {\"type\": \"string\", \"description\": \"Markdown content\"},\n           },\n           \"required\": [\"cell_index\", \"content\"],\n       }\n   },\n   {\n       \"name\": \"get_cells\",\n       \"description\": \"Retrieve current notebook cells and their outputs.\",\n       \"parameters\": {\n           \"type\": \"object\",\n           \"properties\": {\n               \"cell_index_start\": {\"type\": \"integer\", \"description\": \"Start index\", \"default\": 0},\n               \"include_outputs\": {\"type\": \"boolean\", \"description\": \"Include cell outputs\", \"default\": True},\n           },\n           \"required\": [],\n       }\n   },\n]\n\n\n\n\nclass NotebookState:\n\n\n   def __init__(self):\n       self.cells: list[dict] = []\n       self.execution_ns: dict = {\"__builtins__\": __builtins__}\n\n\n   def add_code_cell(self, index: int, code: str) -> dict:\n       cell = {\"type\": \"code\", \"source\": code, \"outputs\": [], \"executed\": False}\n       self.cells.insert(min(index, len(self.cells)), cell)\n       return {\"status\": \"ok\", \"cell_count\": len(self.cells)}\n\n\n   def add_text_cell(self, index: int, content: str) -> dict:\n       cell = {\"type\": \"markdown\", \"source\": content}\n       self.cells.insert(min(index, len(self.cells)), cell)\n       return {\"status\": \"ok\", \"cell_count\": len(self.cells)}\n\n\n   def execute_code(self, code: str) -> dict:\n       stdout_buf = io.StringIO()\n       try:\n           with contextlib.redirect_stdout(stdout_buf):\n               try:\n                   result = eval(code, self.execution_ns)\n                   if result is not None:\n                       return {\"outputs\": [{\"type\": \"result\", \"text\": repr(result)}]}\n               except SyntaxError:\n                   exec(code, self.execution_ns)\n           out = stdout_buf.getvalue()\n           return {\"outputs\": [{\"type\": \"stdout\", \"text\": out}] if out else []}\n       except Exception as e:\n           return {\"outputs\": [{\"type\": \"error\", \"text\": f\"{type(e).__name__}: {e}\"}]}\n\n\n   def get_cells(self, start: int = 0, include_outputs: bool = True) -> dict:\n       return {\"cells\": self.cells[start:], \"total\": len(self.cells)}\n\n\n\n\nclass MCPAgentLoop:\n\n\n   def __init__(self):\n       self.notebook = NotebookState()\n       self.history: list[dict] = []\n       self.max_iterations = 10\n\n\n   def _dispatch_tool(self, name: str, args: dict) -> dict:\n       if name == \"execute_code\":\n           return self.notebook.execute_code(args[\"code\"])\n       elif name == \"add_code_cell\":\n           return self.notebook.add_code_cell(args[\"cell_index\"], args[\"code\"])\n       elif name == \"add_text_cell\":\n           return self.notebook.add_text_cell(args[\"cell_index\"], args[\"content\"])\n       elif name == \"get_cells\":\n           return self.notebook.get_cells(\n               args.get(\"cell_index_start\", 0),\n               args.get(\"include_outputs\", True),\n           )\n       else:\n           return {\"error\": f\"Unknown tool: {name}\"}\n\n\n   def _plan(self, task: str, iteration: int, last_result: dict = None) -> list[dict]:\n       task_lower = task.lower()\n\n\n       if iteration == 0:\n           return [\n               {\"tool\": \"add_text_cell\", \"args\": {\n                   \"cell_index\": 0,\n                   \"content\": f\"# AI-Generated Analysisnn**Task**: {task}nn\"\n                              f\"*Generated by MCP Agent*\"\n               }},\n           ]\n       elif iteration == 1:\n           return [\n               {\"tool\": \"add_code_cell\", \"args\": {\n                   \"cell_index\": 1,\n                   \"code\": \"import randomnimport mathnn\"\n                           \"# Generate sample datan\"\n                           \"random.seed(42)n\"\n                           \"data = [random.gauss(100, 15) for _ in range(500)]n\"\n                           \"print(f'Generated {len(data)} data points')n\"\n                           \"print(f'Sample: {data[:5]}')\"\n               }},\n               {\"tool\": \"execute_code\", \"args\": {\n                   \"code\": \"import randomnimport mathnn\"\n                           \"random.seed(42)n\"\n                           \"data = [random.gauss(100, 15) for _ in range(500)]n\"\n                           \"print(f'Generated {len(data)} data points')n\"\n                           \"print(f'Sample: {[round(x,2) for x in data[:5]]}')\"\n               }},\n           ]\n       elif iteration == 2:\n           return [\n               {\"tool\": \"add_code_cell\", \"args\": {\n                   \"cell_index\": 2,\n                   \"code\": \"# Statistical analysisn\"\n                           \"mean = sum(data) \/ len(data)n\"\n                           \"variance = sum((x - mean)**2 for x in data) \/ len(data)n\"\n                           \"std = variance ** 0.5n\"\n                           \"median = sorted(data)[len(data)\/\/2]n\"\n                           \"print(f'Mean: {mean:.2f}')n\"\n                           \"print(f'Std Dev: {std:.2f}')n\"\n                           \"print(f'Median: {median:.2f}')\"\n               }},\n               {\"tool\": \"execute_code\", \"args\": {\n                   \"code\": \"mean = sum(data) \/ len(data)n\"\n                           \"variance = sum((x - mean)**2 for x in data) \/ len(data)n\"\n                           \"std = variance ** 0.5n\"\n                           \"median = sorted(data)[len(data)\/\/2]n\"\n                           \"print(f'Mean: {mean:.2f}')n\"\n                           \"print(f'Std Dev: {std:.2f}')n\"\n                           \"print(f'Median: {median:.2f}')\"\n               }},\n           ]\n       elif iteration == 3:\n           return [\n               {\"tool\": \"add_text_cell\", \"args\": {\n                   \"cell_index\": 3,\n                   \"content\": \"## Results Summarynn\"\n                              \"The analysis is complete. Key findings are computed above.\"\n                              \"The data follows a normal distribution centered around 100.\"\n               }},\n           ]\n       else:\n           return []\n\n\n   async def run(self, task: str):\n       print(f\"\"\ud83e\udd16\" Agent Task: {task}\")\n       print(\"=\" * 60)\n\n\n       for i in range(self.max_iterations):\n           plan = self._plan(task, i)\n           if not planned:\n               print(f\"n\"\ud83c\udfc1\" Agent finished after {i} iterations\")\n               break\n\n\n           print(f\"n--- Iteration {i+1} ---\")\n\n\n           for step in plan:\n               tool_name = step[\"tool\"]\n               tool_args = step[\"args\"]\n\n\n               print(f\"  \"\ud83d\udd27\" Calling: {tool_name}\")\n               result = self._dispatch_tool(tool_name, tool_args)\n\n\n               self.history.append({\n                   \"iteration\": i,\n                   \"tool\": tool_name,\n                   \"result\": result,\n               })\n\n\n               if \"outputs\" in result:\n                   for out in result[\"outputs\"]:\n                       prefix = \"\"\ud83d\udce4\"\" if out[\"type\"] != \"error\" else \"\"\u26a0\"\"\n                       text = out[\"text\"][:200]\n                       print(f\"     {prefix} {text}\")\n               elif \"status\" in result:\n                   print(f\"     \"\u2705\" {result}\")\n\n\n       print(f\"n\"\ud83d\udcd3\" Final Notebook State:\")\n       print(\"=\" * 60)\n       for i, cell in enumerate(self.notebook.cells):\n           icon = \"\"\ud83d\udcbb\"\" if cell[\"type\"] == \"code\" else \"\"\ud83d\udcdd\"\"\n           source = cell[\"source\"][:60] + (\"...\" if len(cell[\"source\"]) > 60 else \"\")\n           print(f\"  [{i}] {icon} {cell['type']:10s} | {source}\")\n\n\n\n\nagent = MCPAgentLoop()\nasyncio.run(agent.run(\"Analyze a dataset with descriptive statistics\"))\n\n\n\n\nINTEGRATION_TEMPLATE = '''\nimport anthropic\nimport json\n\n\nclient = anthropic.Anthropic()\n\n\ntools = [\n   {\n       \"name\": \"colab-proxy-mcp_add_code_cell\",\n       \"description\": \"Add a Python code cell to the connected Colab notebook\",\n       \"input_schema\": {\n           \"type\": \"object\",\n           \"properties\": {\n               \"cellIndex\": {\"type\": \"integer\"},\n               \"code\": {\"type\": \"string\"},\n               \"language\": {\"type\": \"string\", \"default\": \"python\"},\n           },\n           \"required\": [\"cellIndex\", \"code\"],\n       }\n   },\n   {\n       \"name\": \"colab-proxy-mcp_add_text_cell\",\n       \"description\": \"Add a markdown cell to the connected Colab notebook\",\n       \"input_schema\": {\n           \"type\": \"object\",\n           \"properties\": {\n               \"cellIndex\": {\"type\": \"integer\"},\n               \"content\": {\"type\": \"string\"},\n           },\n           \"required\": [\"cellIndex\", \"content\"],\n       }\n   },\n   {\n       \"name\": \"colab-proxy-mcp_execute_cell\",\n       \"description\": \"Execute a cell in the connected Colab notebook\",\n       \"input_schema\": {\n           \"type\": \"object\",\n           \"properties\": {\n               \"cellIndex\": {\"type\": \"integer\"},\n           },\n           \"required\": [\"cellIndex\"],\n       }\n   },\n   {\n       \"name\": \"colab-proxy-mcp_get_cells\",\n       \"description\": \"Get cells from the connected Colab notebook\",\n       \"input_schema\": {\n           \"type\": \"object\",\n           \"properties\": {\n               \"cellIndexStart\": {\"type\": \"integer\", \"default\": 0},\n               \"includeOutputs\": {\"type\": \"boolean\", \"default\": True},\n           },\n       }\n   },\n   {\n       \"name\": \"runtime_execute_code\",\n       \"description\": \"Execute Python code directly in the Colab kernel (Runtime Mode)\",\n       \"input_schema\": {\n           \"type\": \"object\",\n           \"properties\": {\n               \"code\": {\"type\": \"string\"},\n           },\n           \"required\": [\"code\"],\n       }\n   },\n]\n\n\n\n\ndef run_agent(task: str, max_turns: int = 15):\n   messages = [{\"role\": \"user\", \"content\": task}]\n\n\n   for turn in range(max_turns):\n       response = client.messages.create(\n           model=\"claude-sonnet-4-20250514\",\n           max_tokens=4096,\n           tools=tools,\n           messages=messages,\n           system=\"You are an AI assistant with access to a Google Colab notebook.\"\n                  \"via MCP tools. Build notebooks step by step: add markdown cells \"\n                  \"For documentation, add code cells, then execute them. \"\n                  \"Inspect outputs and fix errors iteratively.\"\n       )\n\n\n       assistant_content = response.content\n       messages.append({\"role\": \"assistant\", \"content\": assistant_content})\n\n\n       if response.stop_reason == \"end_turn\":\n           print(\"Agent finished.\")\n           break\n\n\n       tool_results = []\n       for block in assistant_content:\n           if block.type == \"tool_use\":\n               print(f\"Tool call: {block.name}({json.dumps(block.input)[:100]})\")\n\n\n               result = dispatch_to_mcp_server(block.name, block.input)\n\n\n               tool_results.append({\n                   \"type\": \"tool_result\",\n                   \"tool_use_id\": block.id,\n                   \"content\": json.dumps(result),\n               })\n\n\n       if tool_results:\n           messages.append({\"role\": \"user\", \"content\": tool_results})\n       else:\n           break\n\n\n\n\ndef dispatch_to_mcp_server(tool_name: str, tool_input: dict) -> dict:\n   raise NotImplementedError(\"Use the MCP SDK for real tool dispatch\")\n'''\n\n\nprint(INTEGRATION_TEMPLATE)\nprint(\"n\" + \"=\" * 60)\nprint(\"\"\ud83d\udca1\" The template above shows how to connect a real LLM to colab-mcp.\")\nprint(\"   For Claude Code: just add the MCP config and start chatting!\")\nprint(\"   For custom agents: use the Anthropic SDK with tool_use.\")<\/code><\/pre>\n<\/div>\n<\/div>\n
We build a complete MCPAgentLoop that replicates how real AI agents interact with colab-mcp: it receives a task, plans a sequence of tool calls, dispatches them to a NotebookState manager, inspects outputs, and iterates until the notebook is fully built. We watch the agent run four iterations, which add a markdown title cell, import libraries, generate data, compute descriptive statistics, and write a summary, producing a four-cell notebook entirely through tool calls, with every execution result printed inline. We then print a full-production integration template showing both the zero-code path (a JSON config block for Claude Code or the Gemini CLI) and the custom-agent path (a complete Anthropic API loop with tool definitions, message history management, and tool-result wiring).<\/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\nimport io\nimport contextlib\nimport traceback\nimport uuid\nimport time\nfrom enum import Enum\nfrom dataclasses import dataclass, field\nimport nest_asyncio\nnest_asyncio.apply()\n\n\n@dataclass\nclass KernelOutput:\n   output_type: str\n   text: str = \"\"\n   data: dict = field(default_factory=dict)\n   traceback_lines: list = field(default_factory=list)\n\n\n@dataclass\nclass ExecutionResult:\n   success: bool\n   outputs: list[KernelOutput]\n   execution_count: int\n\n\n\n\nclass ColabRuntimeSimulator:\n\n\n   def __init__(self):\n       self._id = uuid.uuid4()\n       self._execution_count = 0\n       self._namespace: dict = {\"__builtins__\": __builtins__}\n       self._is_started = False\n\n\n   @property\n   def runtime_id(self) -> str:\n       return str(self._id)\n\n\n   async def start(self) -> None:\n       print(f\"   \"\ud83d\udd04\" Initializing runtime {self.runtime_id[:8]}...\")\n       print(f\"   \"\ud83d\udd10\" [Simulated] OAuth2 authentication...\")\n       await asyncio.sleep(0.1)\n       print(f\"   \"\ud83d\udda5\"  [Simulated] Requesting VM assignment...\")\n       await asyncio.sleep(0.1)\n       print(f\"   \"\ud83d\udd0c\" [Simulated] Connecting to Jupyter kernel...\")\n       await asyncio.sleep(0.1)\n       self._is_started = True\n       print(f\"   \"\u2705\" Runtime started!\")\n\n\n   async def execute_code(self, code: str) -> ExecutionResult:\n       if not self._is_started:\n           await self.start()\n\n\n       self._execution_count += 1\n       outputs: list[KernelOutput] = []\n       stdout_buf = io.StringIO()\n       stderr_buf = io.StringIO()\n\n\n       try:\n           with contextlib.redirect_stdout(stdout_buf), \n                contextlib.redirect_stderr(stderr_buf):\n               try:\n                   result = eval(code, self._namespace)\n                   if result is not None:\n                       outputs.append(KernelOutput(\n                           output_type=\"execute_result\",\n                           text=repr(result),\n                           data={\"text\/plain\": repr(result)},\n                       ))\n               except SyntaxError:\n                   exec(code, self._namespace)\n\n\n           stdout_text = stdout_buf.getvalue()\n           if stdout_text:\n               outputs.append(KernelOutput(\n                   output_type=\"stream\",\n                   text=stdout_text,\n               ))\n\n\n           stderr_text = stderr_buf.getvalue()\n           if stderr_text:\n               outputs.append(KernelOutput(\n                   output_type=\"stream\",\n                   text=stderr_text,\n               ))\n\n\n           return ExecutionResult(\n               success=True,\n               outputs=outputs,\n               execution_count=self._execution_count,\n           )\n\n\n       except Exception as e:\n           tb = traceback.format_exc()\n           outputs.append(KernelOutput(\n               output_type=\"error\",\n               text=str(e),\n               traceback_lines=tb.split(\"n\"),\n           ))\n           return ExecutionResult(\n               success=False,\n               outputs=outputs,\n               execution_count=self._execution_count,\n           )\n\n\n   async def stop(self) -> None:\n       if self._is_started:\n           print(f\"   \"\ud83d\uded1\" Unassigning VM for runtime {self.runtime_id[:8]}...\")\n           self._is_started = False\n           print(f\"   \"\u2705\" Runtime stopped and VM released.\")\n\n\n\n\nclass ExecutionStatus(Enum):\n   SUCCESS = \"success\"\n   ERROR = \"error\"\n   TIMEOUT = \"timeout\"\n   RETRYING = \"retrying\"\n\n\n@dataclass\nclass CellExecution:\n   cell_index: int\n   code: str\n   status: ExecutionStatus = ExecutionStatus.SUCCESS\n   output: str = \"\"\n   error: str = \"\"\n   retries: int = 0\n   duration_ms: float = 0.0\n\n\n\n\nclass RobustNotebookOrchestrator:\n\n\n   def __init__(self, max_retries: int = 3, timeout_seconds: float = 30.0):\n       self.max_retries = max_retries\n       self.timeout_seconds = timeout_seconds\n       self.runtime = ColabRuntimeSimulator()\n       self.executions: list[CellExecution] = []\n       self._started = False\n\n\n   async def ensure_started(self):\n       if not self._started:\n           await self.runtime.start()\n           self._started = True\n\n\n   async def execute_with_retry(self, code: str, cell_index: int) -> CellExecution:\n       await self.ensure_started()\n       cell = CellExecution(cell_index=cell_index, code=code)\n\n\n       for attempt in range(self.max_retries + 1):\n           start_time = time.time()\n\n\n           try:\n               result = await asyncio.wait_for(\n                   self.runtime.execute_code(code),\n                   timeout=self.timeout_seconds,\n               )\n               cell.duration_ms = (time.time() - start_time) * 1000\n\n\n               if result.success:\n                   cell.status = ExecutionStatus.SUCCESS\n                   cell.output = \"n\".join(\n                       o.text for o in result.outputs if o.text\n                   )\n                   break\n               else:\n                   error_text = \"n\".join(\n                       o.text for o in result.outputs if o.output_type == \"error\"\n                   )\n                   cell.error = error_text\n\n\n                   if self._is_retryable(error_text) and attempt < self.max_retries:\n                       cell.status = ExecutionStatus.RETRYING\n                       cell.retries = attempt + 1\n                       print(f\"   \"\u267b\"  Retry {attempt + 1}\/{self.max_retries}: {error_text[:60]}\")\n                       await asyncio.sleep(0.5 * (attempt + 1))\n                       continue\n                   else:\n                       cell.status = ExecutionStatus.ERROR\n                       break\n\n\n           except asyncio.TimeoutError:\n               cell.duration_ms = self.timeout_seconds * 1000\n               cell.status = ExecutionStatus.TIMEOUT\n               cell.error = f\"Execution timed out after {self.timeout_seconds}s\"\n               break\n\n\n       self.executions.append(cell)\n       return cell\n\n\n   def _is_retryable(self, error: str) -> bool:\n       retryable_patterns = [\n           \"ConnectionError\", \"TimeoutError\", \"ResourceExhausted\",\n           \"ServiceUnavailable\", \"CUDA out of memory\",\n       ]\n       return any(p.lower() in error.lower() for p in retryable_patterns)\n\n\n   async def execute_notebook(self, cells: list[dict]) -> dict:\n       await self.ensure_started()\n       results = []\n       failed = False\n\n\n       print(\"\"\ud83d\udcd3\" Executing notebook...\")\n       print(\"=\" * 50)\n\n\n       for i, cell in enumerate(cells):\n           if cell.get(\"type\") == \"markdown\":\n               print(f\"  [{i}] \"\ud83d\udcdd\" Markdown: {cell['source'][:50]}...\")\n               results.append({\"index\": i, \"type\": \"markdown\", \"status\": \"ok\"})\n               continue\n\n\n           if failed and not cell.get(\"force_execute\", False):\n               print(f\"  [{i}] \"\u23ed\"  Skipped (previous cell failed)\")\n               results.append({\"index\": i, \"type\": \"code\", \"status\": \"skipped\"})\n               continue\n\n\n           print(f\"  [{i}] \"\ud83d\udcbb\" Executing...\")\n           exec_result = await self.execute_with_retry(cell[\"source\"], i)\n\n\n           icon = {\n               ExecutionStatus.SUCCESS: \"\"\u2705\"\",\n               ExecutionStatus.ERROR: \"\"\u274c\"\",\n               ExecutionStatus.TIMEOUT: \"\"\u23f0\"\",\n               ExecutionStatus.RETRYING: \"\"\u267b\"\",\n           }[exec_result.status]\n\n\n           print(f\"       {icon} {exec_result.status.value} \"\n                 f\"({exec_result.duration_ms:.0f}ms)\"\n                 f\"{f' [{exec_result.retries} retries]' if exec_result.retries else ''}\")\n\n\n           if exec_result.output:\n               for line in exec_result.output.strip().split(\"n\")[:3]:\n                   print(f\"       \"\ud83d\udce4\" {line}\")\n\n\n           if exec_result.status in (ExecutionStatus.ERROR, ExecutionStatus.TIMEOUT):\n               print(f\"       \"\u26a0\"  {exec_result.error[:100]}\")\n               failed = True\n\n\n           results.append({\n               \"index\": i,\n               \"type\": \"code\",\n               \"status\": exec_result.status.value,\n               \"output\": exec_result.output,\n               \"error\": exec_result.error,\n           })\n\n\n       success_count = sum(1 for r in results if r[\"status\"] in (\"ok\", \"success\"))\n       fail_count = sum(1 for r in results if r[\"status\"] in (\"error\", \"timeout\"))\n       skip_count = sum(1 for r in results if r[\"status\"] == \"skipped\")\n\n\n       print(f\"n\"\ud83d\udcca\" Summary: {success_count} passed, {fail_count} failed, {skip_count} skipped\")\n       return {\"results\": results, \"success\": fail_count == 0}\n\n\n   def get_execution_report(self) -> str:\n       lines = [\"Execution Report\", \"=\" * 40]\n       for e in self.executions:\n           lines.append(\n               f\"Cell [{e.cell_index}]: {e.status.value} \"\n               f\"({e.duration_ms:.0f}ms, {e.retries} retries)\"\n           )\n           if e.error:\n               lines.append(f\"  Error: {e.error[:80]}\")\n       return \"n\".join(lines)\n\n\n\n\nasync def advanced_demo():\n   orchestrator = RobustNotebookOrchestrator(max_retries=2, timeout_seconds=5.0)\n\n\n   notebook_cells = [\n       {\"type\": \"markdown\", \"source\": \"# Advanced Analysis\"},\n       {\"type\": \"code\", \"source\": \"x = 42nprint(f'x = {x}')\"},\n       {\"type\": \"code\", \"source\": \"y = x * 2nprint(f'y = x * 2 = {y}')\"},\n       {\"type\": \"code\", \"source\": \"result = x + ynprint(f'Result: {result}')\"},\n       {\"type\": \"code\", \"source\": \"broken_var + 1\"},\n       {\"type\": \"code\", \"source\": \"print('This gets skipped')\"},\n       {\"type\": \"code\", \"source\": \"print('Force executed')\", \"force_execute\": True},\n   ]\n\n\n   result = await orchestrator.execute_notebook(notebook_cells)\n\n\n   print(f\"n{orchestrator.get_execution_report()}\")\n\n\nasyncio.run(advanced_demo())\n\n\n\n\nSUMMARY = \"\"\"\n\u2554\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2557\n\u2551                      Tutorial Complete! \"\ud83c\udf89\"                         \u2551\n\u2560\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2563\n\u2551                                                                    \u2551\n\u2551  What You Learned:                                                 \u2551\n\u2551  \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500                                                 \u2551\n\u2551  \"\u2705\" MCP protocol fundamentals (tools\/list, tools\/call)             \u2551\n\u2551  \"\u2705\" FastMCP framework (how colab-mcp is built)                     \u2551\n\u2551  \"\u2705\" Session Proxy Mode (WebSocket bridge to browser)               \u2551\n\u2551  \"\u2705\" Runtime Mode (direct kernel execution)                         \u2551\n\u2551  \"\u2705\" Full AI agent loop with tool dispatch                          \u2551\n\u2551  \"\u2705\" Production integration with Claude\/GPT-4\/Gemini                \u2551\n\u2551  \"\u2705\" Error handling, retries, and orchestration patterns             \u2551\n\u2551                                                                    \u2551\n\u2551  Quick Start (on your local machine):                              \u2551\n\u2551  \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500                              \u2551\n\u2551  1. pip install uv                                                 \u2551\n\u2551  2. Add to your MCP config:                                        \u2551\n\u2551     {                                                              \u2551\n\u2551       \"mcpServers\": {                                              \u2551\n\u2551         \"colab-proxy-mcp\": {                                       \u2551\n\u2551           \"command\": \"uvx\",                                        \u2551\n\u2551           \"args\": [\"git+https:\/\/github.com\/googlecolab\/colab-mcp\"],\u2551\n\u2551           \"timeout\": 30000                                         \u2551\n\u2551         }                                                          \u2551\n\u2551       }                                                            \u2551\n\u2551     }                                                              \u2551\n\u2551  3. Open a Colab notebook in your browser                          \u2551\n\u2551  4. Tell your agent: \"Build me a data analysis notebook.\"           \u2551\n\u2551                                                                    \u2551\n\u2551  Resources:                                                        \u2551\n\u2551  \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500                                                        \u2551\n\u2551  \"\ud83d\udce6\" Repo:   github.com\/googlecolab\/colab-mcp                      \u2551\n\u2551  \"\ud83d\udcd6\" Docs:   deepwiki.com\/googlecolab\/colab-mcp                    \u2551\n\u2551  \"\ud83d\udcac\" Forum:  GitHub Discussions on the repo                         \u2551\n\u2551  \"\ud83d\udd27\" MCP:    modelcontextprotocol.io                                \u2551\n\u2551                                                                    \u2551\n\u255a\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u255d\n\"\"\"\nprint(SUMMARY)<\/code><\/pre>\n<\/div>\n<\/div>\n
We build a RobustNotebookOrchestrator that adds production-grade resilience on top of the runtime engine: automatic retries with exponential backoff for transient errors like ConnectionError or CUDA out of memory, configurable timeouts via asyncio.wait_for, and dependency-aware cell sequencing that skips downstream cells when an upstream cell fails. We execute a seven-cell notebook that includes three successful computations, one deliberate NameError, one auto-skipped cell, and one force-executed cell, then print a structured execution report with per-cell status, duration, and retry counts. We close with a summary card listing everything we have covered and the exact four steps needed to go from this tutorial to a live colab-mcp deployment on our own machine.<\/p>\n
In conclusion, we now have a working, end-to-end understanding of how colab-mcp turns Google Colab into a programmable workspace for AI agents. We have seen the MCP protocol from both sides, as server authors registering tools and as client code dispatching calls, and we understand why the dual-mode architecture exists: Session Proxy for interactive, browser-visible notebook manipulation, and Runtime for headless, direct kernel execution. We have built the same abstractions the real codebase uses (FastMCP servers, WebSocket bridges with token security, lazy-init resource chains), and we have run them ourselves rather than just reading about them. Most importantly, we have a clear path from this tutorial to real deployment: we take the MCP config JSON, point Claude Code or the Gemini CLI at it, open a Colab notebook, and start issuing natural-language commands that the agent automatically translates into add_code_cell, execute_cell, and get_cells calls. The orchestration patterns from retries, timeouts, and skip-on-failure give us the resilience we need when we move from demos to actual workflows involving large datasets, GPU-accelerated training, or multi-step analyses.<\/p>\n
\n
Check out\u00a0the\u00a0Full Notebook here<\/a>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0120k+ 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 Design a Production-Ready AI Agent That Automates Google Colab Workflows Using Colab-MCP, MCP Tools, FastMCP, and Kernel Execution<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we build an …<\/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-601","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\/601","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=601"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/601\/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=601"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=601"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=601"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}