Enable AI agents (Claude, ChatGPT, custom LLMs) to discover and execute PyExecutor workflows via MCP.
The Model Context Protocol (MCP) is a standardized way for AI agents to discover and call tools/functions. PyExecutor exposes your workflows as MCP tools, allowing any compatible AI model to:
Architecture Flow:
Claude / ChatGPT / Custom LLM
↓
MCP Protocol (WebSocket/Stdio)
↓
PyExecutor MCP Server
↓
REST API → Workflows → Execution Engine
↓
Database, APIs, Scripts, Notifications
# In your PyExecutor backend directory
pip install fastmcp pydantic
# Or add to requirements.txt
fastmcp>=0.1.0
pydantic>=2.0.0# backend/mcp_server.py
from fastmcp import FastMCP
from fastmcp.resources import Resource
from pydantic import BaseModel
import os
import httpx
app = FastMCP("pyexec")
# Get PyExecutor API config
API_URL = os.getenv("PYEXEC_API_URL", "http://localhost:8000")
API_KEY = os.getenv("PYEXEC_API_KEY", "")
class WorkflowInput(BaseModel):
workflow_name: str
context: dict = {}
wait: bool = True
timeout: int = 300
@app.list_tools()
async def list_tools():
"""Discover available workflows as MCP tools"""
async with httpx.AsyncClient() as client:
headers = {"X-API-Key": API_KEY}
response = await client.get(
f"{API_URL}/api/workflows/",
headers=headers
)
workflows = response.json()
tools = []
for workflow in workflows:
tools.append({
"name": f"execute_{workflow['name'].replace('-', '_')}",
"description": workflow.get('description', f"Execute {workflow['name']} workflow"),
"inputSchema": {
"type": "object",
"properties": {
"context": {
"type": "object",
"description": "Context variables for workflow execution",
"additionalProperties": {"type": "string"}
}
},
"required": []
}
})
return tools
@app.call_tool()
async def call_tool(name: str, arguments: dict):
"""Execute a PyExecutor workflow via MCP"""
# Extract workflow name from tool name
workflow_name = name.replace("execute_", "").replace("_", "-")
context = arguments.get("context", {})
async with httpx.AsyncClient() as client:
headers = {"X-API-Key": API_KEY, "Content-Type": "application/json"}
# Execute workflow
response = await client.post(
f"{API_URL}/api/workflows/{workflow_name}/execute/",
json={"context": context},
headers=headers
)
if response.status_code != 200:
return f"Error executing workflow: {response.text}"
job_data = response.json()
job_id = job_data['id']
# Wait for completion
import asyncio
while True:
status_response = await client.get(
f"{API_URL}/api/jobs/{job_id}/",
headers=headers
)
job = status_response.json()
if job['status'] in ['success', 'failed', 'cancelled']:
return {
"status": job['status'],
"output": job.get('output', {}),
"error": job.get('error')
}
await asyncio.sleep(2)
if __name__ == "__main__":
app.run()