server.py•3.17 kB
#!/usr/bin/env python3
"""MCP server for web scraping using newspaper4k."""
import logging
import subprocess
from pathlib import Path
from pydantic import Field
from typing import Annotated
from textwrap import dedent
import time
import dotenv
import os
from agentrun import AgentRun
from fastmcp import FastMCP
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class ComposeService:
def __init__(self, project_dir: str):
self.project_dir = Path(project_dir).resolve()
if not (self.project_dir / "docker-compose.yml").exists():
raise FileNotFoundError(f"No docker-compose.yml in {self.project_dir}")
def __enter__(self):
start_time = time.time()
logger.info(f"Starting docker compose in {self.project_dir}...")
subprocess.run(
["docker", "compose", "up", "-d", "--build"],
cwd=self.project_dir,
check=True
)
elapsed = time.time() - start_time
logger.info(f"Containers started in {elapsed:.2f} seconds")
return self # optional, so you can add helper methods later
def __exit__(self, exc_type, exc_val, exc_tb):
logger.info(f"Stopping docker compose in {self.project_dir}...")
subprocess.run(
["docker", "compose", "down"],
cwd=self.project_dir,
check=True
)
logger.info("Containers stopped.")
# Create FastMCP server instance
mcp = FastMCP(
name="Python Code Executor",
instructions=dedent("""
This server provides Python code execution capabilities.
It can execute Python code in a container.
""")
)
@mcp.tool()
def execute_python_code(
python_code: Annotated[str, Field(description="Python code to execute. Must end with a print statement.")]
) -> str:
"""Execute Python code in a container.
Args:
python_code: Python code to execute. Must end with a print statement.
Returns:
A string containing the result of the code execution.
"""
try:
agentrun_api_dir = os.getenv("AGENTRUN_API_DIR")
dotenv.load_dotenv(os.path.join(agentrun_api_dir, ".env.dev"))
container_name = os.getenv("CONTAINER_NAME")
logger.info("MCP Python Code Executor Server starting...")
with ComposeService(agentrun_api_dir):
runner = AgentRun(container_name=container_name) # container should be running
result = runner.execute_code_in_container(python_code)
return result
except Exception as e:
logger.error(f"Error executing code: {str(e)}")
raise
def main():
import signal
import sys
def signal_handler(signum, frame):
"""Handle shutdown signals."""
logger.info("Shutting down server...")
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
logger.info("MCP Python Code Executor Server starting...")
mcp.run()
if __name__ == "__main__":
main()