Unit 9 · Connect to external tools and data

Handle long-running MCP work

Modern servers may return a task from an ordinary tool call, which Ax can auto-await or expose for input, cancellation, and observation. Legacy task-draft APIs remain compatibility-only.

11 focused minutesNot started

Worked example

See the idea in context

target = AxEventTarget(
    "reindex-flow", invoke,
    mapInput=lambda event, continuation: {
        "taskId": continuation.metadata["taskId"]
        if continuation else event.data["taskId"]
    },
    waitFor=[{"kind": "mcp.task", "value": "taskKey", "metadata": {}}],
)
runtime = AxEventRuntime([
    AxEventRoute("task-start", "wake",
        {"types": ["app.task.started"]}, "reindex-flow"),
    AxEventRoute("task-progress", "observe",
        {"types": ["mcp.progress"]}),
    AxEventRoute("task-resume", "resume",
        {"types": ["mcp.task.status"]}, "reindex-flow"),
], {"targets": [target], "sources": [started, mcp]})
runtime.start()
task_id = client.call_tool("start_reindex", {"scope": "all"})["task"]["taskId"]
target.waitFor[0]["metadata"] = {"taskId": task_id}

Run itIn your own project

pip install axllm

import os
from axllm import ai, ax

llm = ai('openai', api_key=os.environ['OPENAI_API_KEY'])
classify = ax('review:string -> sentiment:class "positive, negative, neutral"')
result = classify.forward(llm, {
    'review': 'Useful and boring in the best way.',
})

print('sentiment:', result['sentiment'])

Set OPENAI_APIKEY in your environment before running provider-backed code.

In the ax repo

From a clone of the ax repo:

npm run example -- python src/examples/python/mcp/native-mcp-tools.py

Active practice

Show that you can use it

Answer 2 in a row to learn this · attempt 1

Keep exploring

See the idea in context

Show that you can use it

Source-backed follow-up