import asyncio
import json
import io
import contextlib
import re
from dataclasses import dataclass
from typing import Callable, Awaitable
import nest_asyncio
nest_asyncio.apply()
TOOL_DEFINITIONS = [
{
"name": "execute_code",
"description": "Execute Python code in the Colab kernel. Returns stdout, results, or errors. State persists between calls."
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Python code to execute"},
},
"required": ["code"],
}
},
{
"title": "add_code_cell",
"description": "Add a code cell to the pocket book at a given index.",
"parameters": {
"sort": "object",
"properties": {
"cell_index": {"sort": "integer", "description": "Place to insert"},
"code": {"sort": "string", "description": "Python code for the cell"},
},
"required": ["cell_index", "code"],
}
},
{
"title": "add_text_cell",
"description": "Add a markdown documentation cell to the pocket book.",
"parameters": {
"sort": "object",
"properties": {
"cell_index": {"sort": "integer", "description": "Place to insert"},
"content material": {"sort": "string", "description": "Markdown content material"},
},
"required": ["cell_index", "content"],
}
},
{
"title": "get_cells",
"description": "Retrieve present pocket book cells and their outputs.",
"parameters": {
"sort": "object",
"properties": {
"cell_index_start": {"sort": "integer", "description": "Begin index", "default": 0},
"include_outputs": {"sort": "boolean", "description": "Embrace cell outputs", "default": True},
},
"required": [],
}
},
]
class NotebookState:
def __init__(self):
self.cells: record[dict] = []
self.execution_ns: dict = {"__builtins__": __builtins__}
def add_code_cell(self, index: int, code: str) -> dict:
cell = {"sort": "code", "supply": code, "outputs": [], "executed": False}
self.cells.insert(min(index, len(self.cells)), cell)
return {"standing": "okay", "cell_count": len(self.cells)}
def add_text_cell(self, index: int, content material: str) -> dict:
cell = {"sort": "markdown", "supply": content material}
self.cells.insert(min(index, len(self.cells)), cell)
return {"standing": "okay", "cell_count": len(self.cells)}
def execute_code(self, code: str) -> dict:
stdout_buf = io.StringIO()
attempt:
with contextlib.redirect_stdout(stdout_buf):
attempt:
consequence = eval(code, self.execution_ns)
if consequence will not be None:
return {"outputs": [{"type": "result", "text": repr(result)}]}
besides SyntaxError:
exec(code, self.execution_ns)
out = stdout_buf.getvalue()
return {"outputs": [{"type": "stdout", "text": out}] if out else []}
besides Exception as e:
return {"outputs": [{"type": "error", "text": f"{type(e).__name__}: {e}"}]}
def get_cells(self, begin: int = 0, include_outputs: bool = True) -> dict:
return {"cells": self.cells[start:], "complete": len(self.cells)}
class MCPAgentLoop:
def __init__(self):
self.pocket book = NotebookState()
self.historical past: record[dict] = []
self.max_iterations = 10
def _dispatch_tool(self, title: str, args: dict) -> dict:
if title == "execute_code":
return self.pocket book.execute_code(args["code"])
elif title == "add_code_cell":
return self.pocket book.add_code_cell(args["cell_index"], args["code"])
elif title == "add_text_cell":
return self.pocket book.add_text_cell(args["cell_index"], args["content"])
elif title == "get_cells":
return self.pocket book.get_cells(
args.get("cell_index_start", 0),
args.get("include_outputs", True),
)
else:
return {"error": f"Unknown instrument: {title}"}
def _plan(self, process: str, iteration: int, last_result: dict = None) -> record[dict]:
task_lower = process.decrease()
if iteration == 0:
return [
{"tool": "add_text_cell", "args": {
"cell_index": 0,
"content": f"# AI-Generated Analysisnn**Task**: {task}nn"
f"*Generated by MCP Agent*"
}},
]
elif iteration == 1:
return [
{"tool": "add_code_cell", "args": {
"cell_index": 1,
"code": "import randomnimport mathnn"
"# Generate sample datan"
"random.seed(42)n"
"data = [random.gauss(100, 15) for _ in range(500)]n"
"print(f'Generated {len(information)} information factors')n"
"print(f'Pattern: {information[:5]}')"
}},
{"instrument": "execute_code", "args": {
"code": "import randomnimport mathnn"
"random.seed(42)n"
"information = [random.gauss(100, 15) for _ in range(500)]n"
"print(f'Generated {len(information)} information factors')n"
"print(f'Pattern: {[round(x,2) for x in data[:5]]}')"
}},
]
elif iteration == 2:
return [
{"tool": "add_code_cell", "args": {
"cell_index": 2,
"code": "# Statistical analysisn"
"mean = sum(data) / len(data)n"
"variance = sum((x - mean)**2 for x in data) / len(data)n"
"std = variance ** 0.5n"
"median = sorted(data)[len(data)//2]n"
"print(f'Imply: {imply:.2f}')n"
"print(f'Std Dev: {std:.2f}')n"
"print(f'Median: {median:.2f}')"
}},
{"instrument": "execute_code", "args": {
"code": "imply = sum(information) / len(information)n"
"variance = sum((x - imply)**2 for x in information) / len(information)n"
"std = variance ** 0.5n"
"median = sorted(information)[len(data)//2]n"
"print(f'Imply: {imply:.2f}')n"
"print(f'Std Dev: {std:.2f}')n"
"print(f'Median: {median:.2f}')"
}},
]
elif iteration == 3:
return [
{"tool": "add_text_cell", "args": {
"cell_index": 3,
"content": "## Results Summarynn"
"The analysis is complete. Key findings are computed above."
"The data follows a normal distribution centered around 100."
}},
]
else:
return []
async def run(self, process: str):
print(f"🤖 Agent Activity: {process}")
print("=" * 60)
for i in vary(self.max_iterations):
plan = self._plan(process, i)
if not deliberate:
print(f"n🏁 Agent completed after {i} iterations")
break
print(f"n--- Iteration {i+1} ---")
for step in plan:
tool_name = step["tool"]
tool_args = step["args"]
print(f" 🔧 Calling: {tool_name}")
consequence = self._dispatch_tool(tool_name, tool_args)
self.historical past.append({
"iteration": i,
"instrument": tool_name,
"consequence": consequence,
})
if "outputs" in consequence:
for out in consequence["outputs"]:
prefix = "📤" if out["type"] != "error" else "⚠️"
textual content = out["text"][:200]
print(f" {prefix} {textual content}")
elif "standing" in consequence:
print(f" ✅ {consequence}")
print(f"n📓 Remaining Pocket book State:")
print("=" * 60)
for i, cell in enumerate(self.pocket book.cells):
icon = "💻" if cell["type"] == "code" else "📝"
supply = cell["source"][:60] + ("..." if len(cell["source"]) > 60 else "")
print(f" [{i}] {icon} {cell['type']:10s} | {supply}")
agent = MCPAgentLoop()
asyncio.run(agent.run("Analyze a dataset with descriptive statistics"))
INTEGRATION_TEMPLATE = '''
import anthropic
import json
shopper = anthropic.Anthropic()
instruments = [
{
"name": "colab-proxy-mcp_add_code_cell",
"description": "Add a Python code cell to the connected Colab notebook",
"input_schema": {
"type": "object",
"properties": {
"cellIndex": {"type": "integer"},
"code": {"type": "string"},
"language": {"type": "string", "default": "python"},
},
"required": ["cellIndex", "code"],
}
},
{
"title": "colab-proxy-mcp_add_text_cell",
"description": "Add a markdown cell to the linked Colab pocket book",
"input_schema": {
"sort": "object",
"properties": {
"cellIndex": {"sort": "integer"},
"content material": {"sort": "string"},
},
"required": ["cellIndex", "content"],
}
},
{
"title": "colab-proxy-mcp_execute_cell",
"description": "Execute a cell within the linked Colab pocket book",
"input_schema": {
"sort": "object",
"properties": {
"cellIndex": {"sort": "integer"},
},
"required": ["cellIndex"],
}
},
{
"title": "colab-proxy-mcp_get_cells",
"description": "Get cells from the linked Colab pocket book",
"input_schema": {
"sort": "object",
"properties": {
"cellIndexStart": {"sort": "integer", "default": 0},
"includeOutputs": {"sort": "boolean", "default": True},
},
}
},
{
"title": "runtime_execute_code",
"description": "Execute Python code instantly within the Colab kernel (Runtime Mode)",
"input_schema": {
"sort": "object",
"properties": {
"code": {"sort": "string"},
},
"required": ["code"],
}
},
]
def run_agent(process: str, max_turns: int = 15):
messages = [{"role": "user", "content": task}]
for flip in vary(max_turns):
response = shopper.messages.create(
mannequin="claude-sonnet-4-20250514",
max_tokens=4096,
instruments=instruments,
messages=messages,
system="You're an AI assistant with entry to a Google Colab pocket book."
"through MCP instruments. Construct notebooks step-by-step: add markdown cells "
"For documentation, add code cells, then execute them. "
"Examine outputs and repair errors iteratively."
)
assistant_content = response.content material
messages.append({"position": "assistant", "content material": assistant_content})
if response.stop_reason == "end_turn":
print("Agent completed.")
break
tool_results = []
for block in assistant_content:
if block.sort == "tool_use":
print(f"Instrument name: {block.title}({json.dumps(block.enter)[:100]})")
consequence = dispatch_to_mcp_server(block.title, block.enter)
tool_results.append({
"sort": "tool_result",
"tool_use_id": block.id,
"content material": json.dumps(consequence),
})
if tool_results:
messages.append({"position": "person", "content material": tool_results})
else:
break
def dispatch_to_mcp_server(tool_name: str, tool_input: dict) -> dict:
increase NotImplementedError("Use the MCP SDK for actual instrument dispatch")
'''
print(INTEGRATION_TEMPLATE)
print("n" + "=" * 60)
print("💡 The template above exhibits methods to join an actual LLM to colab-mcp.")
print(" For Claude Code: simply add the MCP config and begin chatting!")
print(" For customized brokers: use the Anthropic SDK with tool_use.")
