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.

from cortexhub import Cortexhub
cx = Cortexhub(api_key="cxh_key_...")   # or set CORTEXHUB_API_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 with a signed consent_url -- you show that to the user and poll until they decide in CortexHub. Needs both extras: pip install 'cortexhub[ai,mcp]'.

How consent reaches your user. The gateway picks the consent surface from what your MCP client declares at initialize. A UI host that declares the MCP-UI / Tasks extension (e.g. claude.ai) gets consent rendered inline. A plain backend client like the one below declares neither, so the gateway hands it the consent_url fallback shown here -- a signed link you surface to your end user however your app does UI (a page, an email, a Slack DM), then poll cortexhub_get_task for the decision. This is the right pattern for a headless or autonomous agent (there is no host to render inline consent).

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:
                        print("Approve to continue:", task["consent_url"])   # show the end user
                        text = await _await_decision(mcp, task["task_id"])   # wait for the decision

                    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 a signed consent_url.
    def walk(node):
        if isinstance(node, dict):
            t = node.get("mcp_task")
            if isinstance(t, dict) and t.get("consent_url"):
                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.