acp.py.j2

"""ACP handler for {{ agent_name }} — an async OpenAI Agents SDK agent.

Uses the async ACP model with Redis streaming instead of HTTP yields. The
OpenAI Agents SDK run is wrapped in an ``OpenAITurn`` and pushed to the task
stream via ``UnifiedEmitter.auto_send_turn`` — the async delivery path of the
unified harness surface. ``auto_send_turn`` returns a ``TurnResult`` carrying
the accumulated final text and normalized usage.

The agent and its tools are defined inline below so this template stays a
single, self-contained ``acp.py``.
"""

from __future__ import annotations

import os
from typing import List
from datetime import datetime

from dotenv import load_dotenv

load_dotenv()

from agents import Agent, Runner, function_tool, set_tracing_disabled

from agentex.lib import adk
from agentex.lib.types.acp import SendEventParams, CancelTaskParams, CreateTaskParams
from agentex.lib.types.fastacp import AsyncACPConfig
from agentex.lib.types.tracing import SGPTracingProcessorConfig
from agentex.lib.utils.logging import make_logger
from agentex.types.text_content import TextContent
from agentex.lib.utils.model_utils import BaseModel
from agentex.lib.sdk.fastacp.fastacp import FastACP
from agentex.lib.core.harness.emitter import UnifiedEmitter
from agentex.lib.adk import OpenAITurn
from agentex.lib.core.tracing.tracing_processor_manager import add_tracing_processor_config

# Disable the openai-agents SDK's native tracer so it doesn't ship traces to
# api.openai.com using OPENAI_API_KEY (which may be a LiteLLM proxy key).
# SGP tracing below still runs via the Agentex tracing manager.
set_tracing_disabled(True)

logger = make_logger(__name__)

# LiteLLM proxy auth: copy LITELLM_API_KEY to OPENAI_API_KEY for OpenAI client compatibility.
_litellm_key = os.environ.get("LITELLM_API_KEY")
if _litellm_key and not os.environ.get("OPENAI_API_KEY"):
    os.environ["OPENAI_API_KEY"] = _litellm_key

_sgp_api_key = os.environ.get("SGP_API_KEY", "")
_sgp_account_id = os.environ.get("SGP_ACCOUNT_ID", "")
if _sgp_api_key and _sgp_account_id:
    add_tracing_processor_config(
        SGPTracingProcessorConfig(
            sgp_api_key=_sgp_api_key,
            sgp_account_id=_sgp_account_id,
            sgp_base_url=os.environ.get("SGP_CLIENT_BASE_URL", ""),
        )
    )

acp = FastACP.create(
    acp_type="async",
    config=AsyncACPConfig(type="base"),
)

MODEL_NAME = "gpt-4o"
INSTRUCTIONS = """You are a helpful AI assistant with access to tools.

Current date and time: {timestamp}

Guidelines:
- Be concise and helpful
- Use the weather tool when the user asks about the weather
- Always report the real tool output back to the user
"""


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


def create_agent() -> Agent:
    """Build and return the OpenAI Agents SDK agent with the weather tool."""
    return Agent(
        name="{{ agent_name }}",
        model=MODEL_NAME,
        instructions=INSTRUCTIONS.format(timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
        tools=[get_weather],
    )


def get_agent() -> Agent:
    """Build a fresh agent per request so the timestamp in the instructions stays current."""
    return create_agent()


class StateModel(BaseModel):
    """Per-task conversation state persisted between turns."""

    input_list: List[dict]
    turn_number: int


@acp.on_task_create
async def handle_task_create(params: CreateTaskParams):
    logger.info(f"Task created: {params.task.id}")


@acp.on_task_event_send
async def handle_task_event_send(params: SendEventParams):
    """Handle each user message: run the agent and auto-send its turn."""
    agent = get_agent()
    task_id = params.task.id
    agent_id = params.agent.id
    content = params.event.content
    if not isinstance(content, TextContent):
        logger.warning("Ignoring non-text event content (type=%s)", getattr(content, "type", "?"))
        return
    user_message = content.content

    logger.info(f"Processing message for task {task_id}")

    # Echo the user's message into the task history.
    await adk.messages.create(task_id=task_id, content=params.event.content)

    # Load (or create) the persisted conversation history for this task so the
    # agent can see prior turns, then append the new user message.
    task_state = await adk.state.get_by_task_and_agent(task_id=task_id, agent_id=agent_id)
    if task_state is None:
        state = StateModel(input_list=[], turn_number=0)
        task_state = await adk.state.create(task_id=task_id, agent_id=agent_id, state=state)
    else:
        state = StateModel.model_validate(task_state.state)

    state.turn_number += 1
    state.input_list.append({"role": "user", "content": user_message})

    async with adk.tracing.span(
        trace_id=task_id,
        task_id=task_id,
        name="message",
        input={"message": user_message},
        data={"__span_type__": "AGENT_WORKFLOW"},
    ) as turn_span:
        result = Runner.run_streamed(starting_agent=agent, input=state.input_list)
        turn = OpenAITurn(result=result, model=MODEL_NAME)
        emitter = UnifiedEmitter(
            task_id=task_id,
            trace_id=task_id,
            parent_span_id=turn_span.id if turn_span else None,
        )
        turn_result = await emitter.auto_send_turn(turn)

        # Persist the full conversation history (user + assistant + tool calls)
        # so the next turn resumes with complete context.
        state.input_list = result.to_input_list()
        await adk.state.update(
            state_id=task_state.id,
            task_id=task_id,
            agent_id=agent_id,
            state=state,
        )

        if turn_span:
            turn_span.output = {"final_output": turn_result.final_text}


@acp.on_task_cancel
async def handle_task_canceled(params: CancelTaskParams):
    logger.info(f"Task canceled: {params.task.id}")