arthur-ai/arthur-engine/arthur-onboard-instrument

cat .arthur-engine.env 2>/dev/null || echo "(no state file)"

You are an expert Python developer. Instrument the agentic application at: <REPO_PATH>
 
Arthur Engine URL: <ARTHUR_ENGINE_URL>
Arthur Task ID: <ARTHUR_TASK_ID>
 
RULES:
- Never hardcode API keys. Always read from env vars.
- Add to .env (create if missing; ensure .env is gitignored):
    ARTHUR_API_KEY=$ARTHUR_API_KEY
    ARTHUR_BASE_URL=<ARTHUR_ENGINE_URL>
    ARTHUR_TASK_ID=<ARTHUR_TASK_ID>
- Add placeholders to .env.example.
- Smallest possible changes — instrument, don't refactor.
- Print final JSON on the last line: {"success":true,"testsPassed":true,"summary":"<one sentence>"}
 
STEP 1 — ANALYSIS:
- Read requirements.txt / pyproject.toml (note package manager: pip/uv/poetry)
- Find the entry point (main.py, app.py, __main__.py, or similar)
- Identify the LLM framework in use. Match it to one of the supported SDK extras below.
- Check if arthur_observability_sdk already installed (skip STEP 2 if yes)
 
STEP 2 — IMPLEMENTATION:
 
PART A — SDK SETUP:
 
  Supported framework extras and their instrument methods — pick the one that matches:
    openai              → instrument_openai()
    langchain           → instrument_langchain()
    anthropic           → instrument_anthropic()
    crewai              → instrument_crewai()
    autogen             → instrument_autogen()
    autogen-agentchat   → instrument_autogen_agentchat()
    llama-index         → instrument_llama_index()
    bedrock             → instrument_bedrock()
    vertexai            → instrument_vertexai()
    google-genai        → instrument_google_genai()
    google-adk          → instrument_google_adk()
    mistralai           → instrument_mistralai()
    groq                → instrument_groq()
    litellm             → instrument_litellm()
    openai-agents       → instrument_openai_agents()
    claude-agent-sdk    → instrument_claude_agent_sdk()
    haystack            → instrument_haystack()
    dspy                → instrument_dspy()
    smolagents          → instrument_smolagents()
    mcp                 → instrument_mcp()
    (others: ag2, agentminds, agentspec, agno, baml, beeai, cohere,
             guardrails, instructor, ollama, pipecat, portkey, together)
 
  If the framework is not in the list, fall back to the OpenInference instrumentation
  approach instead (see the OpenInference task prompt).
 
  Add "arthur-observability-sdk[<extra>]" to requirements.txt / pyproject.toml.
  Use "arthur-observability-sdk[all]" if unsure or if multiple frameworks are used.
 
  In the entry point, add Arthur initialization (after any existing imports):
    from arthur_observability_sdk import Arthur
    import os
 
    # Arthur raises ValueError if none of task_id / task_name / service_name is given.
    # NOTE: ARTHUR_TASK_ID env var is NOT read automatically — must be passed explicitly.
    # NOTE: ARTHUR_BASE_URL and ARTHUR_API_KEY ARE read automatically from env.
    arthur = Arthur(
        api_key=os.environ.get("ARTHUR_API_KEY"),        # auto-read, but explicit is cleaner
        base_url=os.environ.get("ARTHUR_BASE_URL", "<ARTHUR_ENGINE_URL>"),  # auto-read
        task_id=os.environ.get("ARTHUR_TASK_ID", "<ARTHUR_TASK_ID>"),       # NOT auto-read
        service_name="<app-name>",
        # resource_attributes: arthur.task is set automatically from task_id — don't add it
    )
    arthur.instrument_<framework>()   # call once; patches the framework's HTTP client
 
  At process exit / application teardown, flush pending spans:
    arthur.shutdown()
 
PART B — SESSION + USER CONTEXT (CRITICAL — without this each LLM call is a separate trace):
 
  The preferred way to tag spans with session and user is arthur.attributes(), which
  works as both a context manager and a decorator:
 
    import uuid
    session_id = <derive from app state, or str(uuid.uuid4())>
 
    # Context manager form (use around the full request handler body):
    with arthur.attributes(session_id=session_id,
                           user_id=user_id):  # omit this kwarg if not tracking users
        # all existing processing code here — LLM calls are recorded automatically
 
    # Decorator form (only for functions that serve a single, fixed session — NOT for
    # per-request handlers, because session_id is captured at decoration time, not call time):
    @arthur.attributes(session_id=session_id)
    def handle_request(message):
        ...
 
  Session-only: use arthur.session(session_id) as context manager or decorator.
  User-only:    use arthur.user(user_id)        as context manager or decorator.
 
  For streaming/generator handlers (yield), use explicit OTel attach/detach because
  Python context managers cannot straddle yield points.
  IMPORTANT: use otel_set_value to store session_id in the OTel context (not just as a
  span attribute) — otherwise child auto-instrumented spans won't inherit session.id:
    from opentelemetry import trace, context as otel_ctx
    from opentelemetry.trace import set_span_in_context
    from opentelemetry.context import set_value as otel_set_value
    from openinference.semconv.trace import SpanAttributes, OpenInferenceSpanKindValues
 
    tracer = trace.get_tracer(__name__)
 
    def streaming_handler(message, session_id, ...):
        root_span = tracer.start_span("handler_name")
        root_span.set_attribute(SpanAttributes.OPENINFERENCE_SPAN_KIND,
                                OpenInferenceSpanKindValues.CHAIN.value)
        root_span.set_attribute(SpanAttributes.INPUT_VALUE, message)
        # Build context with root span as parent AND session_id in OTel context
        # so that all child auto-instrumented spans inherit both correctly:
        ctx = otel_set_value(SpanAttributes.SESSION_ID, session_id,
                             context=set_span_in_context(root_span))
        token = otel_ctx.attach(ctx)
        try:
            for chunk in <inner_generator>:
                otel_ctx.detach(token)
                yield chunk
                token = otel_ctx.attach(ctx)
            root_span.set_attribute(SpanAttributes.OUTPUT_VALUE, <final_response>)
        finally:
            root_span.end()
            otel_ctx.detach(token)
 
  For non-streaming handlers that need a root CHAIN span (e.g., to group multiple LLM
  calls into one trace), add an explicit root span inside the arthur.attributes() block:
    from opentelemetry import trace
    from openinference.semconv.trace import SpanAttributes, OpenInferenceSpanKindValues
    import json
 
    tracer = trace.get_tracer(__name__)
 
    with arthur.attributes(session_id=session_id):
        with tracer.start_as_current_span("<handler_name>") as root_span:
            root_span.set_attribute(SpanAttributes.OPENINFERENCE_SPAN_KIND,
                                    OpenInferenceSpanKindValues.CHAIN.value)
            root_span.set_attribute(SpanAttributes.INPUT_VALUE, <user_input>)
            # all existing processing code here
            root_span.set_attribute(SpanAttributes.OUTPUT_VALUE, <response>)
 
PART C — TOOL SPANS (if app uses LLM tool-calling):
  # Uses `tracer` defined in PART B. If not using PART B, add after arthur initialization:
  #   from opentelemetry import trace; tracer = trace.get_tracer(__name__)
    with tracer.start_as_current_span("<tool_name>") as tool_span:
        tool_span.set_attribute(SpanAttributes.OPENINFERENCE_SPAN_KIND,
                                OpenInferenceSpanKindValues.TOOL.value)
        tool_span.set_attribute(SpanAttributes.TOOL_NAME, "<tool_name>")
        tool_span.set_attribute(SpanAttributes.INPUT_VALUE, json.dumps(<tool_input>))
        tool_span.set_attribute(SpanAttributes.INPUT_MIME_TYPE, "application/json")
        result = <execute_tool(tool_input)>
        output_str = json.dumps(result) if not isinstance(result, str) else result
        tool_span.set_attribute(SpanAttributes.OUTPUT_VALUE, output_str)
        tool_span.set_attribute(SpanAttributes.OUTPUT_MIME_TYPE, "application/json")
 
PART D — RETRIEVER SPANS (if app does RAG/vector search):
    with tracer.start_as_current_span("retrieval") as ret_span:
        ret_span.set_attribute(SpanAttributes.OPENINFERENCE_SPAN_KIND,
                               OpenInferenceSpanKindValues.RETRIEVER.value)
        ret_span.set_attribute(SpanAttributes.INPUT_VALUE, <search_query>)
        docs = <execute_retrieval(query)>
        retrieved = []
        for i, doc in enumerate(docs):
            doc_text = <doc_content>
            ret_span.set_attribute(f"retrieval.documents.{i}.document.content", doc_text)
            entry = {"document_content": doc_text}
            if <score_available>:
                score = float(<doc_score>)
                ret_span.set_attribute(f"retrieval.documents.{i}.document.score", score)
                entry["score"] = score
            retrieved.append(entry)
        # REQUIRED: set output.value so retrieved docs appear in Arthur Engine UI
        ret_span.set_attribute(SpanAttributes.OUTPUT_VALUE, json.dumps(retrieved))
 
PART E — AGENT SPANS (if the app has an orchestrator or multi-agent structure):
 
  Span kind guide:
    CHAIN   — entry point or step coordinator (HTTP handler, main workflow function)
    AGENT   — reasoning loop that autonomously decides which tools or agents to call
    TOOL    — a single tool execution invoked by the LLM
    LLM     — a single call to a model API
 
  # Orchestrator node (LLM + tool loop that decides what to do):
  with tracer.start_as_current_span("orchestrate") as agent_span:
      agent_span.set_attribute(SpanAttributes.OPENINFERENCE_SPAN_KIND,
                               OpenInferenceSpanKindValues.AGENT.value)
      agent_span.set_attribute("agent.name", "<name>")
      agent_span.set_attribute(SpanAttributes.INPUT_VALUE, user_input)
      result = run_agent_loop(user_input)
      agent_span.set_attribute(SpanAttributes.OUTPUT_VALUE, result)
 
  # Subagent called by the orchestrator:
  with tracer.start_as_current_span("<subagent_name>") as sub_span:
      sub_span.set_attribute(SpanAttributes.OPENINFERENCE_SPAN_KIND,
                             OpenInferenceSpanKindValues.AGENT.value)
      sub_span.set_attribute("agent.name", "<subagent_name>")
      sub_span.set_attribute(SpanAttributes.INPUT_VALUE, task_input)
      result = run_subagent(task_input)
      sub_span.set_attribute(SpanAttributes.OUTPUT_VALUE, result)
 
PART F — MCP-BASED APPS (if the app calls MCP servers):
 
  Python with arthur-sdk — use the mcp extra for automatic tracing:
    arthur.instrument_mcp()
    # Auto-traces every MCP tool call as a TOOL span with tool.name,
    # mcp.server_name, and mcp.tool_name set automatically.