Nexus Agent SDK for Python

Python SDK for Nexus Agent.

中文 · API reference · Examples

Installation

Python 3.11+.

pip install nexus-agent-sdk-python

Platform wheels include nxs and rg. Supported platforms: Linux, macOS and Windows on x86_64 and ARM64.

export ANTHROPIC_API_KEY="your-api-key"

Custom runtime:

from nexus_agent_sdk import NexusAgentOptions

options = NexusAgentOptions(cli_path="/path/to/nxs")

Quick Start

import asyncio

from nexus_agent_sdk import query


async def main():
    async for message in query(prompt="What is 2 + 2?"):
        print(message)


asyncio.run(main())

Providers

Anthropic Messages

import os

from nexus_agent_sdk import AnthropicProvider, NexusAgentOptions

options = NexusAgentOptions(
    provider=AnthropicProvider(
        api_key=os.environ["ANTHROPIC_API_KEY"],
        base_url="https://api.anthropic.com",
    ),
    model="claude-sonnet-4-6",
)

OpenAI

import os

from nexus_agent_sdk import NexusAgentOptions, OpenAIProvider

options = NexusAgentOptions(
    provider=OpenAIProvider(
        api_key=os.environ["OPENAI_API_KEY"],
        base_url="https://api.openai.com/v1",
        protocol="responses",
    ),
    model=os.environ["OPENAI_MODEL"],
)

protocol accepts chat_completions or responses. Use base_url for compatible services. Provider fields override matching options.env entries. Unset fields inherit environment configuration. Anthropic credentials use api_key or auth_token.

Provider example · Provider API

query()

import asyncio

from nexus_agent_sdk import AssistantMessage, NexusAgentOptions, TextBlock, query


async def main():
    options = NexusAgentOptions(
        system_prompt="You are a code reviewer.",
        cwd="/path/to/project",
        tools=["Read", "Glob", "Grep"],
        allowed_tools=["Read", "Glob", "Grep"],
        max_turns=3,
    )
    async for message in query(prompt="Review this project", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)


asyncio.run(main())

Options

NexusSDKClient

import asyncio

from nexus_agent_sdk import NexusSDKClient


async def main():
    async with NexusSDKClient() as client:
        await client.query("Remember the number 7")
        print((await client.receive_result()).result)

        await client.query("What number did I give you?")
        async for message in client.receive_response():
            print(message)


asyncio.run(main())

Consume async iterable input with receive_messages() through completion before submitting another query.

Client API · Streaming input

Custom Tools

import asyncio

from nexus_agent_sdk import (
    NexusAgentOptions,
    NexusSDKClient,
    create_sdk_mcp_server,
    tool,
)


@tool("add", "Add two numbers", {"a": int, "b": int})
async def add(args):
    return {"content": [{"type": "text", "text": str(args["a"] + args["b"])}]}


async def main():
    server = create_sdk_mcp_server(name="calculator", tools=[add])
    options = NexusAgentOptions(
        mcp_servers={"calculator": server},
        allowed_tools=["mcp__calculator__add"],
    )
    async with NexusSDKClient(options) as client:
        await client.query("Use the calculator to add 12 and 30")
        print((await client.receive_result()).result)


asyncio.run(main())

External MCP Servers

from nexus_agent_sdk import NexusAgentOptions

options = NexusAgentOptions(
    mcp_servers={
        "tools": {
            "type": "stdio",
            "command": "python",
            "args": ["/path/to/mcp_server.py"],
        }
    }
)

Hooks

from nexus_agent_sdk import HookMatcher, NexusAgentOptions


async def log_tool_call(input_data, tool_use_id, context):
    print(input_data["tool_name"], input_data["tool_input"])
    return {}


options = NexusAgentOptions(
    hooks={
        "PreToolUse": [HookMatcher(hooks=[log_tool_call])],
    }
)

Hooks and permission callbacks

Sessions

from nexus_agent_sdk import NexusAgentOptions, SessionStore

store = SessionStore()
for session in store.list_sessions():
    print(session.session_id)

options = NexusAgentOptions(resume="session-id")
fork_options = NexusAgentOptions(resume="session-id", fork_session=True)

Session API

Types

Error Handling

Error types

Development

uv sync --frozen --group dev
uv run python -m compileall -q examples
uv run ruff check .
uv run ruff format --check .
uv run mypy src/nexus_agent_sdk
uv build

Source checkouts require nxs on PATH or cli_path.

Build and release · Changelog

License

Apache-2.0 · Third-party notices