Tool Injection for mcp-run-python: Enable Python code to call back to agent tools with request/response flow #2037
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.