Python Local MCP Server

  • src
  • python_local
import asyncio import sys from io import StringIO import traceback import builtins from mcp.server.models import InitializationOptions import mcp.types as types from mcp.server import NotificationOptions, Server from pydantic import AnyUrl import mcp.server.stdio server = Server("python_interpreter") # Global state for REPL sessions class ReplSession: def __init__(self): self.locals = {"__builtins__": builtins} self.history = [] def execute(self, code: str) -> tuple[str, str]: stdout_capture = StringIO() stderr_capture = StringIO() old_stdout, old_stderr = sys.stdout, sys.stderr sys.stdout, sys.stderr = stdout_capture, stderr_capture try: # Try to evaluate as expression first try: result = eval(code, self.locals) if result is not None: print(repr(result)) except SyntaxError: # If not an expression, execute as statement exec(code, self.locals) except Exception: traceback.print_exc() finally: sys.stdout, sys.stderr = old_stdout, old_stderr return stdout_capture.getvalue(), stderr_capture.getvalue() sessions: dict[str, ReplSession] = {} @server.list_tools() async def handle_list_tools() -> list[types.Tool]: return [ types.Tool( name="python_repl", description="Interactive Python REPL that maintains session state. NOTE THE USER DOES NOT SEE THE STDOUT/STDERR OUTPUT, MAKE SURE TO FORMAT/SUMMARIZEc IT APPROPRIATELY IN YOUR RESPONSE TO THE USER", inputSchema={ "type": "object", "properties": { "code": {"type": "string", "description": "Python code to execute"}, "session_id": {"type": "string", "description": "Session identifier"}, }, "required": ["code", "session_id"], }, ) ] @server.call_tool() async def handle_call_tool( name: str, arguments: dict | None ) -> list[types.TextContent]: if name != "python_repl": raise ValueError(f"Unknown tool: {name}") if not arguments or "code" not in arguments or "session_id" not in arguments: raise ValueError("Missing code or session_id argument") code = arguments["code"] session_id = arguments["session_id"] if session_id not in sessions: sessions[session_id] = ReplSession() session = sessions[session_id] stdout, stderr = session.execute(code) session.history.append({"code": code, "stdout": stdout, "stderr": stderr}) return [types.TextContent(type="text", text=f"NOTE THE USER DOES NOT SEE THIS OUTPUT, MAKE SURE TO FORMAT IT APPROPRIATELY IN YOUR RESPONSE TO THE USER\n\n{stdout}{stderr}")] @server.list_resources() async def handle_list_resources() -> list[types.Resource]: return [ types.Resource( uri=AnyUrl(f"repl://{session_id}/history"), name=f"REPL Session {session_id}", description="REPL session history", mimeType="text/plain", ) for session_id in sessions ] @server.read_resource() async def handle_read_resource(uri: AnyUrl) -> str: if uri.scheme != "repl": raise ValueError(f"Unsupported URI scheme: {uri.scheme}") session_id = uri.host if session_id not in sessions: raise ValueError(f"Session not found: {session_id}") history = sessions[session_id].history return "\n\n".join( f"In [{i}]: {entry['code']}\n" f"Out[{i}]:\n{entry['stdout']}{entry['stderr']}" for i, entry in enumerate(history) ) async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="python_interpreter", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), ) if __name__ == "__main__": asyncio.run(main())