Cómo diseñar un agente de IA listo para producción que automatice los flujos de trabajo de Google Colab utilizando Colab-MCP, MCP Tools, FastMCP y Kernel Execution
importar asyncio importar json importar io importar contextlib importar re desde clases de datos importar clase de datos desde escribir importar Invocable, Aguardable importar 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”]} }, { “name”: “add_code_cell”, “description”: “Agregar una celda de código al cuaderno en un índice determinado.”, “parameters”: { “type”: “object”, “properties”: { “cell_index”: {“type”: “integer”, “description”: “Posición a insertar”}, “code”: {“type”: “string”, “description”: “Código Python para la celda”}, }, “requerido”: [“cell_index”, “code”]} }, { “name”: “add_text_cell”, “description”: “Agregar una celda de documentación de rebajas al cuaderno.”, “parameters”: { “type”: “object”, “properties”: { “cell_index”: {“type”: “integer”, “description”: “Posición para insertar”}, “content”: {“type”: “string”, “description”: “Contenido de Markdown”}, }, “required”: [“cell_index”, “content”]} }, { “name”: “get_cells”, “description”: “Recuperar las celdas actuales del cuaderno y sus resultados.”, “parameters”: { “type”: “object”, “properties”: { “cell_index_start”: {“type”: “integer”, “description”: “Iniciar índice”, “default”: 0}, “include_outputs”: {“type”: “boolean”, “description”: “Incluir salidas de celda”, “predeterminado”: Verdadero}, }, “obligatorio”: []} }, ]clase NotebookState: def __init__(self): self.cells: lista[dict] = []
self.execution_ns: dict = {“__builtins__”: __builtins__} def add_code_cell(self, index: int, code: str) -> dict: cell = {“type”: “código”, “fuente”: código, “salidas”: []”ejecutado”: False} self.cells.insert(min(index, len(self.cells)), cell) return {“status”: “ok”, “cell_count”: len(self.cells)} def add_text_cell(self, index: int, content: str) -> dict: cell = {“type”: “markdown”, “source”: content} self.cells.insert(min(index, len(self.cells)), cell) return {“status”: “ok”, “cell_count”: len(self.cells)} def ejecutar_code(self, código: str) -> dict: stdout_buf = io.StringIO() prueba: con contextlib.redirect_stdout(stdout_buf): prueba: resultado = eval(code, self.execution_ns) si el resultado no es Ninguno: devuelve {“salidas”: [{“type”: “result”, “text”: repr(result)}]} excepto SyntaxError: exec(código, self.execution_ns) out = stdout_buf.getvalue() return {“salidas”: [{“type”: “stdout”, “text”: out}] si fuera otra cosa []} excepto Excepción como e: return {“salidas”: [{“type”: “error”, “text”: f”{type(e).__name__}: {e}”}]} def get_cells(self, start: int = 0, include_outputs: bool = True) -> dict: return {“cells”: self.cells[start:]”total”: len(self.cells)} clase MCPAgentLoop: def __init__(self): self.notebook = NotebookState() self.history: lista[dict] = []
self.max_iterations = 10 def _dispatch_tool(self, nombre: str, args: dict) -> dict: if nombre == “execute_code”: devuelve self.notebook.execute_code(args[“code”]) nombre elif == “add_code_cell”: devuelve self.notebook.add_code_cell(args[“cell_index”]argumentos[“code”]) nombre elif == “add_text_cell”: devuelve self.notebook.add_text_cell(args[“cell_index”]argumentos[“content”]) elif nombre == “get_cells”: return self.notebook.get_cells( args.get(“cell_index_start”, 0), args.get(“include_outputs”, True), ) else: return {“error”: f”Herramienta desconocida: {nombre}”} def _plan(self, tarea: str, iteración: int, last_result: dict = Ninguno) -> lista[dict]: task_lower = task.lower() si iteración == 0: retorno [
{“tool”: “add_text_cell”, “args”: {
“cell_index”: 0,
“content”: f”# AI-Generated Analysis\n\n**Task**: {task}\n\n”
f”*Generated by MCP Agent*”
}},
]
iteración elif == 1: retorno [
{“tool”: “add_code_cell”, “args”: {
“cell_index”: 1,
“code”: “import random\nimport math\n\n”
“# Generate sample data\n”
“random.seed(42)\n”
“data = [random.gauss(100, 15) for _ in range(500)]\n” “print(f’Generado {len(datos)} puntos de datos’)\n” “print(f’Muestra: {datos[:5]}’)” }}, {“tool”: “execute_code”, “args”: { “code”: “importar aleatorio\nimportar matemáticas\n\n” “random.seed(42)\n” “datos = [random.gauss(100, 15) for _ in range(500)]\n” “print(f’Generado {len(datos)} puntos de datos’)\n” “print(f’Muestra: {[round(x,2) for x in data[:5]]}’)” }}, ]elif iteración == 2: retorno [
{“tool”: “add_code_cell”, “args”: {
“cell_index”: 2,
“code”: “# Statistical analysis\n”
“mean = sum(data) / len(data)\n”
“variance = sum((x – mean)**2 for x in data) / len(data)\n”
“std = variance ** 0.5\n”
“median = sorted(data)[len(data)//2]\n” “print(f’Media: {media:.2f}’)\n” “print(f’Std Dev: {std:.2f}’)\n” “print(f’Mediana: {mediana:.2f}’)” }}, {“tool”: “execute_code”, “args”: { “code”: “media = suma(datos) / len(datos)\n” “varianza = suma((x – media)**2 para x en datos) / len(datos)\n” “std = varianza ** 0.5\n” “mediana = ordenado(datos)[len(data)//2]\n” “print(f’Media: {media:.2f}’)\n” “print(f’Std Dev: {std:.2f}’)\n” “print(f’Mediana: {mediana:.2f}’)” }}, ]iteración elif == 3: retorno [
{“tool”: “add_text_cell”, “args”: {
“cell_index”: 3,
“content”: “## Results Summary\n\n”
“The analysis is complete. Key findings are computed above.”
“The data follows a normal distribution centered around 100.”
}},
]
más: regresar []

async def run(self, task: str): print(f”🤖 Tarea del agente: {task}”) print(“=” * 60) for i in range(self.max_iterations): plan = self._plan(task, i) si no está planificado: print(f”\n🏁 El agente finalizó después de {i} iteraciones”) break print(f”\n— Iteración {i+1} —“) para el paso del plan: nombre_herramienta = paso[“tool”]
tool_args = paso[“args”]

print(f” 🔧 Llamando a: {nombre_herramienta}”) resultado = self._dispatch_tool(nombre_herramienta, argumentos_herramienta) self.history.append({ “iteración”: i, “herramienta”: nombre_herramienta, “resultado”: resultado, }) si “salidas” en el resultado: para salida en el resultado[“outputs”]: prefijo = “📤” si está fuera[“type”] != “error” else “⚠️” texto = fuera[“text”][:200]

print(f” {prefijo} {texto}”) elif “estado” en resultado: print(f” ✅ {resultado}”) print(f”\n📓 Estado final del cuaderno:”) print(“=” * 60) para i, celda en enumerar(self.notebook.cells): icono = “💻” si celda[“type”] == “código” más “📝” fuente = celda[“source”][:60] + (“…” si len(celda[“source”]) > 60 más “”) imprimir(f” [{i}] {icono} {celda[‘type’]:10s} | {fuente}”) agente = MCPAgentLoop() asyncio.run(agent.run(“Analizar un conjunto de datos con estadísticas descriptivas”)) INTEGRATION_TEMPLATE = ”’ importar antrópico importar cliente json = anthropic.Anthropic() herramientas = [
{
“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”]} }, { “name”: “colab-proxy-mcp_add_text_cell”, “description”: “Agregar una celda de rebajas al cuaderno Colab conectado”, “input_schema”: { “type”: “object”, “properties”: { “cellIndex”: {“type”: “integer”}, “content”: {“type”: “string”}, }, “required”: [“cellIndex”, “content”]} }, { “name”: “colab-proxy-mcp_execute_cell”, “description”: “Ejecutar una celda en el cuaderno Colab conectado”, “input_schema”: { “type”: “object”, “properties”: { “cellIndex”: {“type”: “integer”}, }, “required”: [“cellIndex”]} }, { “name”: “colab-proxy-mcp_get_cells”, “description”: “Obtener celdas del cuaderno Colab conectado”, “input_schema”: { “type”: “object”, “properties”: { “cellIndexStart”: {“type”: “integer”, “default”: 0}, “includeOutputs”: {“type”: “boolean”, “default”: True}, }, } }, { “name”: “runtime_execute_code”, “description”: “Ejecutar código Python directamente en el kernel de Colab (modo de ejecución)”, “input_schema”: { “type”: “object”, “properties”: { “code”: {“type”: “string”}, }, “required”: [“code”]} }, ]def run_agent(tarea: str, max_turns: int = 15): mensajes = [{“role”: “user”, “content”: task}]

para turn in range(max_turns): respuesta = client.messages.create( model=”claude-sonnet-4-20250514″, max_tokens=4096, tools=tools, message=messages, system=”Eres un asistente de IA con acceso a una computadora portátil de Google Colab.” “a través de las herramientas MCP. Cree computadoras portátiles paso a paso: agregue celdas de rebajas ” “Para documentación, agregue celdas de código y luego ejecútelas. ” “Inspeccione las salidas y corregir errores de forma iterativa.” ) asistente_content = respuesta.content mensajes.append({“role”: “asistente”, “content”: asistente_content}) if respuesta.stop_reason == “end_turn”: print(“Agente terminado.”) break tool_results = []
para bloque en Assistant_content: if block.type == “tool_use”: print(f”Llamada a herramienta: {block.name}({json.dumps(block.input)[:100]})”) resultado = despacho_to_mcp_server(block.name, block.input) tool_results.append({ “type”: “tool_result”, “tool_use_id”: block.id, “content”: json.dumps(resultado), }) if tool_results: mensajes.append({“role”: “usuario”, “content”: tool_results}) else: break def despacho_to_mcp_server(tool_name: str, tool_input: dict) -> dict: rise NotImplementedError(“Use el SDK de MCP para el envío de herramientas reales”) ”’ print(INTEGRATION_TEMPLATE) print(“\n” + “=” * 60) print(“💡 La plantilla anterior muestra cómo conectar un LLM real a colab-mcp.”) print(” Para Claude Code: simplemente agregue el MCP config y comience a chatear!”) print(” Para agentes personalizados: utilice el SDK de Anthropic con tool_use.”)