#!/usr/bin/env python3
import asyncio
import subprocess
import sys
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio
# Create the MCP server
app = Server("curl-mcp-server")
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List available tools."""
return [
Tool(
name="curl",
description="Execute curl commands with full network access. Supports all curl options and flags.",
inputSchema={
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The complete curl command to execute (e.g., 'curl -k https://example.com')"
}
},
"required": ["command"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""Execute the requested tool."""
if name != "curl":
raise ValueError(f"Unknown tool: {name}")
command = arguments.get("command", "")
if not command.strip():
return [TextContent(type="text", text="Error: Command cannot be empty")]
# Parse the command - split by spaces but handle quotes
import shlex
try:
cmd_parts = shlex.split(command)
except ValueError as e:
return [TextContent(type="text", text=f"Error parsing command: {str(e)}")]
# Ensure the command starts with 'curl'
if not cmd_parts or cmd_parts[0] != "curl":
return [TextContent(type="text", text="Error: Command must start with 'curl'")]
try:
# Execute the curl command
result = subprocess.run(
cmd_parts,
capture_output=True,
text=True,
timeout=300 # 5 minute timeout
)
# Combine stdout and stderr
output = ""
if result.stdout:
output += result.stdout
if result.stderr:
if output:
output += "\n--- STDERR ---\n"
output += result.stderr
if not output:
output = f"Command completed with exit code {result.returncode} (no output)"
return [TextContent(type="text", text=output)]
except subprocess.TimeoutExpired:
return [TextContent(type="text", text="Error: Command timed out after 5 minutes")]
except Exception as e:
return [TextContent(type="text", text=f"Error executing command: {str(e)}")]
async def main():
"""Run the MCP server using stdio transport."""
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
