MCP Adapter¶
The MCP adapter (functualize-mcp) exposes functualize jobs as MCP tools for external AI agents (Claude, Goose, Cursor, etc.). Jobs become discoverable and invocable without code changes.
Quick Start¶
pip install "functualize[cli]" functualize-mcp
# Expose jobs via stdio transport (default)
func mcp serve
# Or via HTTP+SSE
func mcp serve --http --port 8080
How It Works¶
- MCP adapter discovers all visible jobs from the functualize registry
- Jobs are translated into MCP tools using their docstring, config model, and metadata
- External AI agents call
discover_jobs,get_job_schema, andrun_job - Results are returned as structured MCP responses
graph LR
Agent["AI Agent"] -->|"MCP protocol"| Adapter["functualize-mcp"]
Adapter -->|"app.execute()"| Engine["Execution Engine"]
Engine --> Jobs["Job Functions"]
Available MCP Tools¶
Core Tools¶
| Tool | Description |
|---|---|
discover_jobs |
List all visible jobs with name, description, and tags |
get_job_schema(name) |
Get full JSON Schema of a job's config model |
run_job(name, config?) |
Execute a job synchronously and return results |
run_job_async(name, config?) |
Start a job and return an execution ID |
get_execution_status(id) |
Poll async execution state |
Workflow Tools¶
| Tool | Description |
|---|---|
list_workflows(workflow_name?, state?, blocked_on?) |
Survey scopes, with filters |
get_workflow_state(id) |
Graph, results, position, pending gates |
answer_gate(values, workflow_id?, gate?, …) |
Record input for a gate |
get_gate_draft(workflow_id?, gate?) |
What is supplied, missing and invalid |
resume_workflow(id?, input?, gate?, retry_epilogue?) |
Advance the walk |
call_gate_tool(id, tool, args?) |
Run a tool the gate offers |
cancel_workflow(id) |
Cancel — terminal |
purge_workflows(state?, older_than_days?) |
Delete finished scopes |
answer_gaterecords.resume_workflowadvances. One meaning each. Until these split, every tool here was a deposit and nothing an agent could call advanced a blocked run — the only continuation was re-invoking the job from a shell, which an agent over MCP cannot do.
Each tool maps to a func builtin workflow verb, takes the same identifiers,
and returns the same shapes. That is held by a parity test that enumerates both
surfaces, not by this table.
Addressing. answer_gate takes workflow_id and gate, each optional
when unambiguous. resume_workflow's id may be omitted when exactly one scope
can be advanced. Ambiguity is never guessed: several candidates are listed and
the call fails.
Task Tools (when Tasks domain active)¶
| Tool | Description |
|---|---|
add_task(title, linked_to?) |
Create a new task |
list_tasks(filter?, status?) |
Query tasks |
update_task(id, status?, notes?) |
Update task status |
plan_tasks(tasks) |
Replace entire task list |
History Tools (when State domain active)¶
| Tool | Description |
|---|---|
get_job_history(name?, limit?) |
Query execution history |
get_execution_detail(id) |
Get full execution record |
Job-to-Tool Translation¶
Jobs are automatically translated to MCP tools:
- Tool name → job name
- Tool description → first paragraph of job's
__doc__ - Input schema → JSON Schema from Pydantic config model
- Annotations → from
@jobtags - Examples → from
@jobexamples
A job under a JOB_GROUP is served under its full dotted name — probe.echo,
not echo — matching func mcp schema and func mcp tools. Dotted names
are legal MCP tool names (SEP-986 allows [A-Za-z0-9._-]), so an agent calls
the job by the same name functualize uses everywhere else.
Controlling Visibility¶
@job(visibility="external") # Exposed via MCP (default)
def deploy(...): ...
@job(visibility="internal") # Hidden from MCP
def _helper(...): ...
Configuration filtering — set in the [mcp] section of your config files:
[mcp]
include_tags = ["ai", "safe"] # only expose jobs with one of these tags
exclude_jobs = ["internal-job"] # hide specific jobs
enable_management = true # expose the multi-server management tools
Schema Export¶
Export job schemas in multiple formats for AI agent knowledge bases:
func mcp schema --format json # MCP tool definition format
func mcp schema --format markdown # Parameters table per job
func mcp schema --format openai # OpenAI function calling format
func mcp schema --format typescript # TypeScript type definitions
func mcp tools # List exposed tools (no server)
What an agent learns about a job¶
builtin info --job <name> --json and the MCP tool list render the same
payload, so what an agent can find out is the same on both surfaces.
Alongside the callable facts — parameters, inputSchema, dependencies,
requires_tty — it carries the four descriptive fields @job declares:
from functualize.job import job
@job(
category="release",
tags=["deploy", "production"],
examples=["func deploy --target staging"],
extra_description="Requires the release role. Rolls back on failure.",
)
def deploy(target: str) -> None:
"""Deploy to the given target."""
{
"name": "deploy",
"summary": "Deploy to the given target.",
"category": "release",
"tags": ["deploy", "production"],
"examples": ["func deploy --target staging"],
"extra_description": "Requires the release role. Rolls back on failure."
}
These survive discovery and the cache, and this payload used to drop them — which made the one direction that matters impossible: an agent that had found a job could not walk from it to the judgment explaining when to use it.
A job declared by convention renders the same shape, not a different one:
tags and examples come back as [], category and extra_description as
null. A consumer never has to branch on which kind of job it is looking at.
Multi-Server Management¶
Manage MCP servers for multiple functualize projects:
func mcp start ./project-a --name api --port 8080
func mcp start ./project-b --name worker --port 8081
func mcp list # Show running servers
func mcp stop api # Stop by name
func mcp stop --all # Stop all
With enable_management = true in [mcp] config, management is exposed as MCP tools themselves:
[mcp]
enable_management = true
# Exposes: mcp_start_server, mcp_list_servers, mcp_stop_server, mcp_get_server_tools
AI_OUTBOUND Gate Strategy¶
When functualize-mcp is installed, workflows can pause and wait for an external AI agent. Declare a Gate with the ai_outbound strategy:
from functualize.workflow import Gate
Gate(name="review", awaits=ReviewInput, tools=[search_docs], strategy="ai_outbound")
The workflow pauses, becomes visible via list_workflows(), and resumes when the AI agent calls resume_workflow(id, input) — which now genuinely advances the walk rather than only recording the input. ai_outbound always blocks — that is the mechanism, so no resolver is registered for it.
tools names the jobs the agent may call while resolving the gate — a permission enforced at MCP dispatch, not a hint. See Gates.
The same dispatch is available imperatively, without declaring a workflow: