Skip to content

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

  1. MCP adapter discovers all visible jobs from the functualize registry
  2. Jobs are translated into MCP tools using their docstring, config model, and metadata
  3. External AI agents call discover_jobs, get_job_schema, and run_job
  4. 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"]
Hold "Alt" / "Option" to enable pan & zoom

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_gate records. resume_workflow advances. 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 @job tags
  • Examples → from @job examples

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:

jobs/deploy.py
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."""
func builtin info --job deploy --json
{
  "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:

rc.invoke(review_job, awaits_input=ReviewInput, force_gate=True,
          gate_strategy="ai_outbound")