AgenTrux Agent SDK

Beta — the API may change before the 1.0 release.

A toolkit for using AgenTrux event publish/read from any AI agent across many agent frameworks. Works with OpenAI, Anthropic (Claude), LangChain, CrewAI, and any LLM that supports function calling.

← Back to Plugins

Install

Requires Python 3.10 or later.

pip install agentrux-agent-tools

Quick Start

1. Create the toolkit

import asyncio
from agentrux_agent_tools import AgenTruxToolkit

async def main():
    toolkit = await AgenTruxToolkit.create(
        base_url="https://api.agentrux.com",
        client_id="crd_your-credential-id",
        client_secret="aks_your-client-secret",
    )

The client_id (crd_) and client_secret (aks_) come from redeeming an Activation Code (POST /auth/redeem-activation-code) in the Console. Using environment variables:

export AGENTRUX_BASE_URL=https://api.agentrux.com
export AGENTRUX_CLIENT_ID=crd_your-credential-id
export AGENTRUX_CLIENT_SECRET=aks_your-client-secret

    toolkit = await AgenTruxToolkit.create()  # reads from environment

2. Get tool definitions

    # OpenAI function-calling format
    tools = toolkit.get_tools()

    # Anthropic tool_use format
    tools = toolkit.get_tools_anthropic()

3. Execute a tool call from the LLM

    result = await toolkit.execute("publish_event", {
        "topic_id": "550e8400-e29b-41d4-a716-446655440000",
        "event_type": "chat.message",
        "payload": {"text": "Hello from the agent!"},
    })
    print(result)  # JSON string containing the event_id

OpenAI Example

import openai
from agentrux_agent_tools import AgenTruxToolkit

async def agent_loop():
    toolkit = await AgenTruxToolkit.create()
    client = openai.AsyncOpenAI()

    messages = [{"role": "user", "content": "Publish a greeting event"}]

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=toolkit.get_tools(),
    )

    for tool_call in response.choices[0].message.tool_calls or []:
        import json
        args = json.loads(tool_call.function.arguments)
        result = await toolkit.execute(tool_call.function.name, args)
        messages.append(response.choices[0].message)
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": result,
        })

    await toolkit.close()

Claude (Anthropic) Example

import anthropic
from agentrux_agent_tools import AgenTruxToolkit

async def agent_loop():
    toolkit = await AgenTruxToolkit.create()
    client = anthropic.AsyncAnthropic()

    response = await client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        tools=toolkit.get_tools_anthropic(),
        messages=[{"role": "user", "content": "List recent events from my topic"}],
    )

    for block in response.content:
        if block.type == "tool_use":
            result = await toolkit.execute(block.name, block.input)
            # return the result as a tool_result...

    await toolkit.close()

Generic Agent Loop

from agentrux_agent_tools import AgenTruxToolkit

async def generic_agent(llm_call, user_prompt: str):
    """Works with any LLM that supports function calling."""
    async with await AgenTruxToolkit.create() as toolkit:
        tools = toolkit.get_tools()
        messages = [{"role": "user", "content": user_prompt}]

        while True:
            response = await llm_call(messages, tools=tools)

            if not response.tool_calls:
                return response.text

            for call in response.tool_calls:
                result = await toolkit.execute(call.name, call.arguments)
                messages.append({"role": "tool", "content": result})

Available Tools

Environment Variables

Use Cases

Add AgenTrux tools to an OpenAI agent

Give an OpenAI agent the ability to send and receive events.

Setup:

• Topic: agent-workspace (read + write)
• Script: openai-agent

Code:

toolkit = await AgenTruxToolkit.create()
tools = toolkit.get_tools()  # OpenAI format

response = await openai_client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Check for new tasks and process them"}],
    tools=tools,
)

Multi-agent coordination

Multiple agents share a single topic to coordinate work. One writes tasks while others process them.

Setup:

• Topic: task-queue (read + write for all agents)
• Script A: task-publisher (writes tasks)
• Script B: worker-1 (reads and processes tasks)
• Script C: worker-2 (reads and processes tasks)

Event-driven agent loop

Build an agent that continuously watches for new events and responds to them.

Setup:

• Topic A: requests (read) — incoming requests
• Topic B: responses (write) — outgoing responses
• Script: responder-agent

Code:

async with await AgenTruxToolkit.create() as toolkit:
    while True:
        result = await toolkit.execute("wait_for_event", {
            "topic_id": REQUEST_TOPIC,
            "timeout_seconds": 300,
        })
        if result:
            # process and respond
            await toolkit.execute("publish_event", {
                "topic_id": RESPONSE_TOPIC,
                "event_type": "response",
                "payload": {"answer": process(result)},
            })

License

MIT