Code Mode: Tool Injection for mcp-run-python: Enable Python code to call back to agent tools with request/response flow #2037
Open
0 of 2 issues completed
Description
Description
Feature Request: Tool Injection for mcp-run-python
Currently, mcp-run-python executes Python code in complete isolation. This feature request proposes enabling tool injection through MCP notifications, allowing sandboxed Python code to call back to the parent agent's tools while maintaining complete security isolation.
Demo of how code might look when using the feature
# Parent agent setup from pydantic_ai import Agent from pydantic_ai.mcp import MCPServerStdio agent = Agent('claude-3-5-haiku-latest', mcp_servers=[mcp_run_python_server], tools=[web_search_tool, database_query_tool, send_email_tool]) # User query result = await agent.run(""" Find recent AI breakthroughs relevant to our enterprise customers and send them personalized email updates. """) # LLM generates Python code with tool calls (parameters decided by LLM): """ # This runs in isolated Pyodide sandbox but can call back to agent tools recent_news = call_tool("web_search", query="AI breakthroughs enterprise 2025", max_results=20, time_filter="1month") # Python execution blocks here until web_search completes and returns results # LLM determines appropriate SQL query based on user intent customers = call_tool("database_query", sql="SELECT email, company, industry FROM customers WHERE tier='enterprise'") # Python logic processes results relevant_articles = [] for article in recent_news: if any(keyword in article['title'].lower() for keyword in ['enterprise', 'business', 'scalability']): relevant_articles.append(article) # Generate personalized emails with error handling for customer in customers: try: industry_articles = [a for a in relevant_articles if customer['industry'].lower() in a['content'].lower()] if industry_articles: email_body = f"Hi {customer['company']}, here are {len(industry_articles)} AI developments..." call_tool("send_email", to=customer['email'], subject=f"AI Updates for {customer['industry']}", body=email_body) except ToolCallError as e: print(f"Failed to send email to {customer['email']}: {e}") print(f"Processing complete") """Technical Implementation
1. Complete Request/Response Flow
sequenceDiagram participant Python as Python (Pyodide) participant MCP as MCP Server (deno) participant Agent as Agent Client Python->>MCP: call_tool("web_search", query="...") Note over MCP: Generate request_id, store pending call MCP->>Agent: tool_call_request(id="req_123", tool="web_search", args={...}) Note over Agent: Execute web_search tool Agent->>MCP: tool_call_response(id="req_123", result=[...]) Note over MCP: Resume Python execution with result MCP->>Python: return result Note over Python: Continue execution with tool result 2. Enhanced MCP Tool Schema
{ "name": "run_python_code", "inputSchema": { "type": "object", "properties": { "python_code": {"type": "string"}, "available_tools": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"}, "description": {"type": "string"}, "parameters": {"type": "object"} } }, "default": [] } } } }3. MCP Notification Messages
// Tool call request (MCP Server → Agent) interface ToolCallRequest { jsonrpc: "2.0"; method: "notifications/tool_call_request"; params: { requestId: string; toolName: string; arguments: Record<string, any>; }; } // Tool call response (Agent → MCP Server) interface ToolCallResponse { jsonrpc: "2.0"; method: "notifications/tool_call_response"; params: { requestId: string; result?: any; error?: string; }; }4. Implementation in MCP Server (deno)
// Handle pending tool calls with request/response pairing const pendingToolCalls = new Map<string, {resolve: Function, reject: Function}>(); async function callTool(toolName: string, args: any): Promise<any> { const requestId = generateId(); // Send request to agent const request = { jsonrpc: "2.0", method: "notifications/tool_call_request", params: { requestId, toolName, arguments: args } }; // Create promise that resolves when response arrives const responsePromise = new Promise((resolve, reject) => { pendingToolCalls.set(requestId, { resolve, reject }); // 30 second timeout setTimeout(() => { pendingToolCalls.delete(requestId); reject(new Error(`Tool call timeout: ${toolName}`)); }, 30000); }); await sendMessage(request); return responsePromise; // Python execution blocks here } // Handle responses from agent function handleToolResponse(response: ToolCallResponse) { const pending = pendingToolCalls.get(response.params.requestId); if (pending) { pendingToolCalls.delete(response.params.requestId); if (response.params.error) { pending.reject(new Error(response.params.error)); } else { pending.resolve(response.params.result); } } }5. Injected Python Functions
# Available in Pyodide globals during execution def call_tool(tool_name: str, **kwargs): """Call an available agent tool and wait for response""" try: # This calls the deno layer which handles MCP communication result = _internal_call_tool(tool_name, kwargs) return result except Exception as e: raise ToolCallError(f"Tool call '{tool_name}' failed: {str(e)}") def call_mcp_tool(server_name: str, tool_name: str, **kwargs): """Call a tool from an MCP server""" return _internal_call_mcp_tool(server_name, tool_name, kwargs) class ToolCallError(Exception): """Raised when a tool call fails""" passReferences
- HuggingFace's
smolagents(link) implements this pattern with their CodeAgent, where tools are exposed as Python functions during code execution.
Note:
- Samuel did mention this during his PyCon talk, but I couldn't locate an existing issue or PR for this feature. If this has already been discussed elsewhere, please feel free to link to the relevant discussion.