cortexhub

The CortexHub platform SDK. One API key points your existing agent at two governed planes:

• the LLM router (inference) - OpenAI-compatible, so any framework works, with routing, caching, compression, governance, and per-agent metering; and
• the MCP gateway (tools + brain + files) - governed, consent-gated, injection-defended.

You don't rewrite your agent. You change a base URL and drop in one key.

Documentation

Install

pip install cortexhub          # zero-dependency core (URLs + headers)
pip install 'cortexhub[ai]'    # + a ready OpenAI-compatible client via cx.llm
pip install 'cortexhub[mcp]'   # + the official MCP client for cx.mcp() sessions

The key

Onboard an external AI employee in the CortexHub dashboard, grant it capabilities, and mint an API key bound to that agent (Settings -> API keys, or the agent's page). The cxh_key_... is that agent's identity - every call is attributed, governed, and metered to it. No agent argument needed.

import cortexhub
cx = cortexhub.init("cxh_key_...")      # or set CORTEXHUB_API_KEY

# equivalently:
from cortexhub import Cortexhub
cx = Cortexhub(api_key="cxh_key_...")

Run a session (recommended)

Your agent runs in your own runtime, but every run should show up in CortexHub with the same visibility as a co-worker on CortexHub's runtime -- one Session in the Sessions tab, with its Trace, Spans, cost, and user attribution.

Use one cx.session(...) handle for the whole run. Pass the identity only your app knows: a stable session id (your id for the conversation or autonomous run) and the end user it acts for. The handle's .llm (inference) and .mcp (tools) both land in that one session.

s = cx.session(
    agent="support-bot",
    mcp_session_id="conv-42",       # your stable id -> one CortexHub session
    end_user_subject="user_42",     # who it acts for (None for an autonomous run)
)

# Inference (OpenAI-compatible). model="cortexhub/auto" lets CortexHub route per
# turn; or pass any model enabled on the platform.
s.llm.chat.completions.create(model="cortexhub/auto", messages=[{"role": "user", "content": "hi"}])

# Tools + brain + files, in the SAME session:
s.mcp.url       # https://mcp.cortexhub.ai/v1/mcp
s.mcp.headers   # wire into your MCP transport (see "Tools" below)

s.session_id    # "conv-42" -- log or forward it to correlate

Reuse the same handle across the run's turns so they group. An autonomous agent (e.g. a cron job) does the same with a per-run id and no end_user_subject (extra for .llm: cortexhub[ai]).

Inference without a session

For a quick, standalone call, cx.llm is a ready OpenAI-compatible client and cx.ai gives the raw base_url + api_key any framework needs. These are NOT grouped into a session (no Sessions row) -- prefer cx.session(...) above when you want the run visible in CortexHub.

cx.llm.chat.completions.create(model="cortexhub/auto", messages=[...])
cx.llm.responses.create(model="cortexhub/auto", input="hi")
cx.llm.models.list()

cx.ai.base_url   # e.g. https://api.cortexhub.ai/v1  (override with CORTEXHUB_ROUTER_URL)
cx.ai.api_key    # your cxh_key_...

Use your existing framework

The router speaks the OpenAI wire protocol, so every framework integrates the same way: point its OpenAI-compatible provider at cx.ai.base_url, pass cx.ai.api_key, and choose a model.

base, key = cx.ai.base_url, cx.ai.api_key

# OpenAI SDK / plain
from openai import OpenAI
OpenAI(base_url=base, api_key=key).chat.completions.create(model="cortexhub/auto", messages=[...])

# LangChain / LangGraph
from langchain_openai import ChatOpenAI
ChatOpenAI(base_url=base, api_key=key, model="cortexhub/auto")

# CrewAI (LiteLLM under the hood)
from crewai import LLM
LLM(model="openai/cortexhub/auto", base_url=base, api_key=key)

# AutoGen
from autogen_ext.models.openai import OpenAIChatCompletionClient
OpenAIChatCompletionClient(model="cortexhub/auto", base_url=base, api_key=key)

# Pydantic AI
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider
OpenAIModel("cortexhub/auto", provider=OpenAIProvider(base_url=base, api_key=key))

Anthropic SDK: the router is OpenAI-compatible, not Anthropic-Messages, so point the OpenAI client at CortexHub with a claude-* model (we route to Claude) rather than the Anthropic SDK.

Tools, brain, and files (MCP gateway)

s.mcp (the session handle above) is the MCP connection your agent's loop drives. Any MCP-capable framework registers it as a server; the cortexhub_* tools (governed toolkit calls, brain recall/learn, files search) then appear automatically -- and, on the session handle, group with that run's LLM turns.

conn = s.mcp                 # from cx.session(...) -> same session as .llm
conn.url                     # https://mcp.cortexhub.ai/v1/mcp
conn.headers                 # wire into your MCP transport

cx.mcp(agent=..., end_user_subject=...) is the standalone form when you only need tools (no session grouping).

A full turn: reason, act, and approvals

One session, end to end: the model reasons with s.llm, acts through the governed cortexhub_* tools on s.mcp, and when a governed action needs the end user's approval it comes back parked, carrying an approval_id and the presentation (headline / sections / summary) of what will happen. Needs both extras: pip install 'cortexhub[ai,mcp]'.

How consent reaches your user. Your agent is headless, so you own the consent UI. When a call parks, read the approval_id + presentation from the result, show it to your end user however your app does UI, and when they decide submit it through the SDK:

s.decide(approval_id, "approve", note="ok by me")   # or "deny"

s.decide records the decision as that end user's consent -- no CortexHub account for the end user is required -- bound to the same agent + end_user_subject this session runs as. Then poll cortexhub_get_task (below): an approved call runs server-side and its result comes back. If you'd rather not build any UI, the parked result also carries a signed consent_url you can hand the user to approve in a browser instead. (A UI host that declares the MCP-UI extension, e.g. claude.ai, gets consent rendered inline and never reaches this path.)

import asyncio, json
from cortexhub import Cortexhub
from mcp import ClientSession

cx = Cortexhub(api_key="cxh_key_...")

async def run_turn(user_text: str) -> str:
    # One session for this end user's conversation -> LLM + tools share it.
    s = cx.session(agent="support-bot", mcp_session_id="conv-42", end_user_subject="user_42")

    # s.mcp.open() opens the transport across mcp client versions.
    async with s.mcp.open() as (read, write):
        async with ClientSession(read, write) as mcp:
            await mcp.initialize()

            # Offer the governed cortexhub_* tools to the model.
            listed = await mcp.list_tools()
            tools = [{
                "type": "function",
                "function": {"name": t.name, "description": t.description or "",
                             "parameters": t.inputSchema or {"type": "object", "properties": {}}},
            } for t in listed.tools]

            messages = [{"role": "user", "content": user_text}]
            for _ in range(8):                              # a few reason -> act rounds
                msg = s.llm.chat.completions.create(
                    model="cortexhub/auto", messages=messages, tools=tools,
                ).choices[0].message
                if not msg.tool_calls:
                    return msg.content or ""                # final answer

                messages.append(msg.model_dump())
                for call in msg.tool_calls:
                    args = json.loads(call.function.arguments or "{}")
                    result = await mcp.call_tool(call.function.name, args)
                    text = _text(result)

                    task = _parked_task(result)             # governed action -> needs approval?
                    if task:
                        # Show task["presentation"] in YOUR UI and get the user's
                        # decision. Here we auto-approve to keep the example runnable.
                        s.decide(task["approval_id"], "approve", note="demo auto-approve")
                        text = await _await_decision(mcp, task["task_id"])   # resume + read result

                    messages.append({"role": "tool", "tool_call_id": call.id, "content": text})
    return ""

# --- small helpers over the MCP result ---
def _text(result) -> str:
    return "\n".join(b.text for b in result.content if getattr(b, "text", None)) or "{}"

def _payload(result):
    sc = getattr(result, "structuredContent", None)
    if sc:
        return sc
    try:
        return json.loads(_text(result))
    except json.JSONDecodeError:
        return {}

def _parked_task(result):
    # A parked approval comes back as an `mcp_task` carrying an approval_id (and a
    # signed consent_url fallback). The presentation you passed to the tool call
    # rides alongside it, so you can render your own consent UI.
    def walk(node):
        if isinstance(node, dict):
            t = node.get("mcp_task")
            if isinstance(t, dict) and t.get("approval_id"):
                if "presentation" in node and "presentation" not in t:
                    t = {**t, "presentation": node["presentation"]}
                return t
            for v in node.values():
                if (found := walk(v)):
                    return found
        elif isinstance(node, list):
            for item in node:
                if (found := walk(item)):
                    return found
        return None
    return walk(_payload(result))

async def _await_decision(mcp, task_id: str) -> str:
    # Poll until the human decides in CortexHub; an approval runs the action.
    for _ in range(600):                                    # ~10 min at 1s waits
        res = await mcp.call_tool("cortexhub_get_task", {"task_id": task_id, "wait_ms": 1000})
        if _payload(res).get("terminal"):
            return _text(res)
    return "approval timed out"

print(asyncio.run(run_turn("Email the Q3 report to the finance team")))

The whole turn is one CortexHub session (conv-42): open its Sessions tab to see the model step, each tool call, and the approval as a single trace.

Authentication modes

• API key (api_key=cxh_key_..., or CORTEXHUB_API_KEY) - backends whose end users do not have CortexHub accounts. Attest the end user per call with end_user_subject. Required for cx.ai / cx.llm.
• MCP OAuth client (client_id= / client_secret=) - for interactive CortexHub users, MCP gateway only.

The MCP gateway also accepts MCP OAuth from third-party clients (Claude, Cursor) independently of this SDK.