NexusLLM Python SDK

Official asynchronous Python SDK for the Centralized LLM Platform (NexusLLM).

Installation

Install using uv (recommended) or pip:

uv add nexus-llm
# or
pip install nexus-llm

Building a LangChain/LangGraph agent on top of NexusLLM? Install the langchain extra to get ChatCentralizedLLM/NexusEmbeddings (see LangChain / LangGraph Integration below):

pip install "nexus-llm[langchain]"

Quick Start

Simple Chat Completion

import asyncio
from nexus_llm import AIClient

async def main():
    # Automatically reads NEXUS_API_KEY and NEXUS_BASE_URL env variables
    client = AIClient(api_key="your-nexus-api-key")

    response = await client.chat.create(
        agent_id="my-custom-agent",
        messages=[
            {"role": "user", "content": "Hello! Introduce yourself."}
        ]
    )

    print(response.choices[0].content)
    await client.close()

asyncio.run(main())

Streaming Chat

Passing stream=True targets the gateway's dedicated Server-Sent-Events endpoint (/chat/stream) under the hood, rather than the plain JSON /chat endpoint — you don't need to do anything differently, the example below just works.

import asyncio
from nexus_llm import AIClient

async def main():
    client = AIClient(api_key="your-nexus-api-key")

    stream = await client.chat.create(
        agent_id="my-custom-agent",
        messages=[
            {"role": "user", "content": "Write a 3-paragraph story."}
        ],
        stream=True
    )

    async for chunk in stream:
        print(chunk.content, end="", flush=True)

    await client.close()

asyncio.run(main())

Tool Calling

tools accepts raw OpenAI function-calling JSON schema — the format every major agent framework already produces, so it needs no translation on your end. The response's tool_calls mirrors LangChain's own ToolCall shape. When streaming, tool-call arguments arrive incrementally as chunk.tool_call_chunks (partial JSON fragments, keyed by index) rather than as one complete call.

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get the current weather for a city.",
        "parameters": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
}]

response = await client.chat.create(
    agent_id="my-custom-agent",
    messages=[{"role": "user", "content": "What's the weather in Delhi?"}],
    tools=tools,
    tool_choice="auto",
)

for call in response.choices[0].tool_calls or []:
    print(call["name"], call["args"])  # e.g. get_weather {'city': 'Delhi'}

# Feed the executed tool's result back in a follow-up call:
messages = [
    {"role": "user", "content": "What's the weather in Delhi?"},
    {"role": "assistant", "content": None, "tool_calls": response.choices[0].tool_calls},
    {"role": "tool", "content": "72F and sunny", "tool_call_id": response.choices[0].tool_calls[0]["id"]},
]

Not every model/provider combination supports tool calling yet — the gateway returns 424 for one that isn't verified rather than silently ignoring tools. Check supports_tools on an entry from client.models/the admin model garden before wiring up a new agent.

Reasoning Content

Models that expose their reasoning/thinking (Claude's extended thinking, OpenAI's o-series, DeepSeek) surface it as a separate reasoning field on the choice/chunk — never mixed into content — so you can log or display it distinctly from the final answer:

response = await client.chat.create(agent_id="my-custom-agent", messages=[...])
print(response.choices[0].reasoning)  # None if the model doesn't support it
print(response.choices[0].content)

LangChain / LangGraph Integration

Install the langchain extra (pip install "nexus-llm[langchain]"), then swap ChatOpenAI(...) for ChatCentralizedLLM(...) in any LangGraph node — graph topology, tools, memory, and checkpointers are unaffected, since they only depend on the BaseChatModel interface:

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from nexus_llm import AIClient
from nexus_llm.integrations.langchain import ChatCentralizedLLM, NexusEmbeddings

client = AIClient(api_key="your-nexus-api-key")
llm = ChatCentralizedLLM(agent_id="my-custom-agent", client=client)

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"72F and sunny in {city}"

agent = create_react_agent(llm, tools=[get_weather])
result = await agent.ainvoke({"messages": [{"role": "user", "content": "Weather in Delhi?"}]})

Every call — including the tool-calling round trips LangGraph drives internally — still goes through the agent's configured provider/model, budget enforcement, and trace logging on the gateway; the framework has no awareness a gateway sits underneath. NexusEmbeddings wraps client.embeddings behind LangChain's Embeddings interface for RAG-style retrievers/vectorstores.

File Management

Upload files, list them, fetch a presigned download URL, or delete them:

import asyncio
from nexus_llm import AIClient

async def main():
    client = AIClient(api_key="your-nexus-api-key")

    with open("dataset.csv", "rb") as f:
        uploaded = await client.files.upload(f, filename="dataset.csv", purpose="fine-tune")

    files = await client.files.list()
    for f in files:
        print(f.file_id, f.filename, f.size_bytes)

    downloadable = await client.files.get(uploaded.file_id)
    print(downloadable.download_url)

    await client.files.delete(uploaded.file_id)

    await client.close()

asyncio.run(main())

Agent Configuration

Configure the provider/model routing for an agent, and read back its current configuration:

import asyncio
from nexus_llm import AIClient

async def main():
    client = AIClient(api_key="your-nexus-api-key")

    config = await client.agents.configure(
        agent_id="my-custom-agent",
        provider="anthropic",
        model="claude-sonnet-4-5",
        fallback_provider="openai",
        fallback_model="gpt-4o",
        system_prompt="You are a helpful assistant.",
        cache_enabled=True,
    )
    print(config.provider, config.model)

    current = await client.agents.get_config("my-custom-agent")
    print(current.cache_enabled)

    await client.close()

asyncio.run(main())

Error Handling

Errors from the API are mapped to specific custom exception classes:

from nexus_llm import AIClient
from nexus_llm.exceptions import AuthenticationError, RateLimitError, ValidationError

async def run():
    client = AIClient(api_key="invalid-key")
    try:
        await client.chat.create(
            agent_id="agent-1",
            messages=[{"role": "user", "content": "hi"}]
        )
    except AuthenticationError:
        print("API Key invalid or revoked.")
    except RateLimitError:
        print("Rate limits exceeded. Exponential backoff has completed retries.")
    except ValidationError as e:
        print(f"Validation failed: {e.message}")

Configuration Options

When initializing the client, the following parameters are accepted: