Python

Build multi-agent workflows with the OpenAI Agents SDK and send traces to Keywords AI.

This integration is only for Agent tracing. If you are looking for the OpenAI integration with the AI gateway, please see the OpenAI integration.

Give us a star on GitHub!

The OpenAI Agents SDK is a lightweight yet powerful framework for building multi-agent workflows.

Core concepts:

1. Agents: LLMs configured with instructions, tools, guardrails, and handoffs
2. Handoffs: Allow agents to transfer control to other agents for specific tasks
3. Guardrails: Configurable safety checks for input and output validation
4. Tracing: Built-in tracking of agent runs, allowing you to view, debug and optimize your workflows

Quickstart

Install the OpenAI Agents SDK and the Keywords AI exporter, then configure the exporter to send traces from the Agents SDK to Keywords AI.

Prerequisites

pip install openai-agents

pip install keywordsai-exporter-openai-agents

Use the specific endpoint for the OpenAI Agents SDK.

https://api.keywordsai.co/api/openai/v1/traces/ingest

If you are on the enterprise platform, please use the enterprise endpoint plus the suffix.

Hello World example

from dotenv import load_dotenv

load_dotenv(override=True)
import pytest
# ==========copy paste below==========
import asyncio
import os
from agents import Agent, Runner
from agents.tracing import set_trace_processors, trace
from keywordsai_exporter_openai_agents import KeywordsAITraceProcessor

set_trace_processors(
    [
        KeywordsAITraceProcessor(
            api_key=os.getenv("KEYWORDSAI_API_KEY"),
            endpoint=os.getenv("KEYWORDSAI_OAIA_TRACING_ENDPOINT"),
        ),
    ]
)


@pytest.mark.asyncio
async def test_main():
    agent = Agent(
        name="Assistant",
        instructions="You only respond in haikus.",
    )

    with trace("Hello world test"):
        result = await Runner.run(agent, "Tell me about recursion in programming.")
        print(result.final_output)
    # Function calls itself,
    # Looping in smaller pieces,
    # Endless by design.


if __name__ == "__main__":
    asyncio.run(test_main())

Handoffs example

from agents import Agent, Runner
import asyncio
from keywordsai_exporter_openai_agents import KeywordsAITraceProcessor
from agents.tracing import set_trace_processors
import os

set_trace_processors(
    [
        KeywordsAITraceProcessor(
            api_key=os.getenv("KEYWORDSAI_API_KEY"),
            endpoint=os.getenv("KEYWORDSAI_OAIA_TRACING_ENDPOINT"),
        ),
    ]
)

spanish_agent = Agent(
    name="Spanish agent",
    instructions="You only speak Spanish.",
)

english_agent = Agent(
    name="English agent",
    instructions="You only speak English",
)

triage_agent = Agent(
    name="Triage agent",
    instructions="Handoff to the appropriate agent based on the language of the request.",
    handoffs=[spanish_agent, english_agent],
)


async def main():
    result = await Runner.run(triage_agent, input="Hola, ¿cómo estás?")
    print(result.final_output)
    # ¡Hola! Estoy bien, gracias por preguntar. ¿Y tú, cómo estás?


if __name__ == "__main__":
    asyncio.run(main())

Functions example

import asyncio

from agents import Agent, Runner, function_tool
from keywordsai_exporter_openai_agents import KeywordsAITraceProcessor
from agents.tracing import set_trace_processors
import os

set_trace_processors(
    [
        KeywordsAITraceProcessor(
            api_key=os.getenv("KEYWORDSAI_API_KEY"),
            endpoint=os.getenv("KEYWORDSAI_OAIA_TRACING_ENDPOINT"),
        ),
    ]
)


@function_tool
def get_weather(city: str) -> str:
    return f"The weather in {city} is sunny."


agent = Agent(
    name="Hello world",
    instructions="You are a helpful agent.",
    tools=[get_weather],
)


async def main():
    result = await Runner.run(agent, input="What's the weather in Tokyo?")
    print(result.final_output)
    # The weather in Tokyo is sunny.


if __name__ == "__main__":
    asyncio.run(main())