MCP | Ax

These Python examples are real runnable files. Edit the source file first; this page is rebuilt from the checked-in example and its metadata header.

Python Native MCP Tools

Attaches a live MCP client directly to AxGen without a lossy function adapter.

Provider: openai
Env: OPENAI_API_KEY, OPENAI_APIKEY, MCP_URL
Level: beginner
Run: npm run example -- python src/examples/python/mcp/native-mcp-tools.py
Source: src/examples/python/mcp/native-mcp-tools.py

Python

import os

from axllm import AxMCPClient, AxMCPStreamableHTTPTransport, ai, ax

api_key = os.getenv("OPENAI_API_KEY") or os.getenv("OPENAI_APIKEY")
endpoint = os.getenv("MCP_URL")
if not api_key or not endpoint:
    raise SystemExit("Set OPENAI_API_KEY and MCP_URL.")

# The repo's demo MCP server runs on http://127.0.0.1; any other endpoint keeps
# the default SSRF protection (https only, no local hosts).
local = endpoint.startswith("http://127.0.0.1")
transport = AxMCPStreamableHTTPTransport(
    endpoint,
    {"ssrfProtection": {"requireHttps": not local, "allowLocalhost": local, "allowPrivateNetworks": local}},
)
mcp = AxMCPClient(transport, {"namespace": "inventory"})
llm = ai("openai", api_key=api_key, model="gpt-5.4-mini")
program = ax(
    'request:string -> answer:string "Use the inventory MCP tool."',
    {"mcp": mcp},
)
try:
    catalog = mcp.inspect_catalog()
    print({
        "tools": [tool["name"] for tool in catalog["tools"]],
        "resources": [
            {"name": resource["name"], "uri": resource["uri"]}
            for resource in catalog["resources"]
        ],
        "resourceTemplates": catalog["resourceTemplates"],
    })
    print(program.forward(llm, {"request": "Reindex inventory."}))
finally:
    mcp.close()

Python MCP Resource Wake

Subscribes over real Streamable HTTP and lets AxEventRuntime wake an authenticated Agent automatically.

Provider: openai
Env: OPENAI_API_KEY, OPENAI_APIKEY, AX_MCP_ENDPOINT
Level: intermediate
Run: npm run example -- python src/examples/python/mcp/resource-wake-agent.py
Source: src/examples/python/mcp/resource-wake-agent.py

Python

import os
import threading
import urllib.request

from axllm import (
    AxEventRoute,
    AxEventRuntime,
    AxEventTarget,
    AxMCPClient,
    AxMCPEventSource,
    AxMCPStreamableHTTPTransport,
    agent,
    ai,
)
from axllm.runtime_quickjs import AxQuickJsCodeRuntime

api_key = os.getenv("OPENAI_API_KEY") or os.getenv("OPENAI_APIKEY")
if not api_key:
    raise SystemExit("Set OPENAI_API_KEY.")

endpoint = os.environ.get("AX_MCP_ENDPOINT")
if not endpoint:
    raise SystemExit("Set AX_MCP_ENDPOINT to a Streamable HTTP MCP server.")
transport = AxMCPStreamableHTTPTransport(
    endpoint,
    {
        "ssrfProtection": {
            "requireHttps": not endpoint.startswith("http://127.0.0.1"),
            "allowLocalhost": endpoint.startswith("http://127.0.0.1"),
            "allowPrivateNetworks": endpoint.startswith("http://127.0.0.1"),
        }
    },
)
client = AxMCPClient(transport, {"namespace": "inventory"})
source = AxMCPEventSource(
    client,
    "inventory",
    identity_scope="tenant:demo",
    trust="authenticated",
    resource_subscriptions="all",
)
llm = ai("openai", api_key=api_key, model="gpt-5.4-mini")
program = agent("uri:string -> summary:string", {"runtime": {"language": "JavaScript"}})
completed = threading.Event()


def invoke(input, _context):
    output = program.forward(llm, input, {"runtime": AxQuickJsCodeRuntime()})
    print(output)
    completed.set()
    return output


target = AxEventTarget(
    "inventory-agent",
    invoke,
    mapInput=lambda event, _continuation: {"uri": event.data["uri"]},
    retrySafety="idempotent",
)
runtime = AxEventRuntime(
    [
        AxEventRoute(
            "resource-wake",
            "wake",
            {"types": ["mcp.resource.updated"]},
            "inventory-agent",
            True,
        )
    ],
    {"targets": [target], "sources": [source]},
)
runtime.start()
if os.getenv("AX_MCP_DEMO_AUTO") == "1":
    urllib.request.urlopen(
        urllib.request.Request(
            endpoint.replace("/mcp", "/control/resource"), data=b"", method="POST"
        )
    ).close()
if not completed.wait(60):
    raise RuntimeError("Timed out waiting for an MCP resource notification")
runtime.close()
client.close()

Python MCP Task Continuation

Creates an owned continuation and resumes an AxFlow from real MCP progress and terminal task notifications.

Provider: openai
Env: OPENAI_API_KEY, OPENAI_APIKEY, AX_MCP_ENDPOINT
Level: advanced
Run: npm run example -- python src/examples/python/mcp/task-resume-flow.py
Source: src/examples/python/mcp/task-resume-flow.py

Python

import os
import threading
import urllib.request

from axllm import (
    AxEventEnvelope,
    AxEventRoute,
    AxEventRuntime,
    AxEventTarget,
    AxMCPClient,
    AxMCPEventSource,
    AxMCPStreamableHTTPTransport,
    AxPushEventSource,
    ai,
    ax,
    flow,
)

api_key = os.getenv("OPENAI_API_KEY") or os.getenv("OPENAI_APIKEY")
if not api_key:
    raise SystemExit("Set OPENAI_API_KEY.")

endpoint = os.environ.get("AX_MCP_ENDPOINT")
if not endpoint:
    raise SystemExit("Set AX_MCP_ENDPOINT to a Streamable HTTP MCP server.")
transport = AxMCPStreamableHTTPTransport(
    endpoint,
    {
        "ssrfProtection": {
            "requireHttps": not endpoint.startswith("http://127.0.0.1"),
            "allowLocalhost": endpoint.startswith("http://127.0.0.1"),
            "allowPrivateNetworks": endpoint.startswith("http://127.0.0.1"),
        }
    },
)
client = AxMCPClient(transport, {"namespace": "inventory"})
mcp = AxMCPEventSource(
    client, "inventory", identity_scope="tenant:demo", trust="authenticated"
)
started = AxPushEventSource("task-started")
llm = ai("openai", api_key=api_key, model="gpt-5.4-mini")
step = ax("taskId:string -> status:string")
program = (
    flow({"id": "reindex-flow"}).execute("status", step).returns({"status": "status"})
)
completed = threading.Event()
calls = 0


def invoke(input, _context):
    global calls
    output = program.forward(llm, input)
    assert output.get("status"), output
    calls += 1
    print(output)
    if calls >= 2:
        completed.set()
    return output


target = AxEventTarget(
    "reindex-flow",
    invoke,
    mapInput=lambda event, continuation: {
        "taskId": (
            continuation.metadata["taskId"] if continuation else event.data["taskId"]
        )
    },
    retrySafety="idempotent",
    waitFor=[{"kind": "mcp.task", "value": "taskKey", "metadata": {"taskId": "42"}}],
)
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()
# A modern server answers with a task; a legacy server returns it inside a
# complete result.
outcome = client.call_tool_outcome("start_reindex", {"scope": "all"})
task = outcome["task"] if outcome["kind"] == "task" else outcome["result"].get("task")
assert task is not None, "start_reindex did not start an MCP task"
task_id = task["taskId"]
target.waitFor[0]["metadata"] = {"taskId": task_id}
started.publish(
    AxEventEnvelope(
        "task-start",
        "app://tasks",
        "app.task.started",
        {"taskId": task_id, "taskKey": f"inventory:{task_id}"},
    ),
    identity_scope="tenant:demo",
    trust="authenticated",
)
print(f"Task {task_id} is waiting for a terminal MCP notification.")
if os.getenv("AX_MCP_DEMO_AUTO") == "1":
    urllib.request.urlopen(
        urllib.request.Request(
            endpoint.replace("/mcp", "/control/task/complete"), data=b"", method="POST"
        )
    ).close()
if not completed.wait(60):
    raise RuntimeError("Timed out waiting for the MCP task continuation")
runtime.close()
client.close()