EvalView Adapters

Adapters are the bridge between EvalView and your AI agent. EvalView includes dedicated adapters for LangGraph, CrewAI, OpenAI (Responses API), Anthropic Claude, HuggingFace, Ollama, MCP servers, and generic HTTP/streaming APIs. You can also build custom adapters for proprietary frameworks.

Adapters connect EvalView to your AI agent's API. EvalView includes adapters for common patterns and makes it easy to build custom ones.

Built-in Adapters

HTTP Adapter

For standard REST APIs that return a complete JSON response.

Use when:

• Your agent returns a full response in one request
• No streaming involved
• Response contains all steps/tools in a single payload

Configuration:

# .evalview/config.yaml
adapter: http
endpoint: http://localhost:3000/api/agent
timeout: 30.0
headers:
  Authorization: Bearer your-api-key
  Content-Type: application/json

Expected Response Format:

{
  "session_id": "session-123",
  "output": "Final agent response",
  "steps": [
    {
      "id": "step-1",
      "name": "Tool name",
      "tool": "tool_identifier",
      "parameters": {"arg": "value"},
      "output": {"result": "data"},
      "success": true,
      "latency": 234,
      "cost": 0.001,
      "tokens": 150
    }
  ],
  "cost": 0.025,
  "tokens": 1250
}

HuggingFace Adapter

For Gradio-based agents hosted on HuggingFace Spaces.

Use when:

• Your agent is hosted on HuggingFace Spaces
• Your agent uses Gradio for the interface
• You want zero-config connection to HF Spaces

Configuration:

# .evalview/config.yaml
adapter: huggingface  # or 'hf' or 'gradio'

# Supports multiple URL formats:
endpoint: username/my-space
# OR: https://huggingface.co/spaces/username/my-space
# OR: https://username-my-space.hf.space

timeout: 120  # Longer timeout for AI inference

Environment Variables:

# HuggingFace token (optional for public Spaces)
export HF_TOKEN="hf_your_token_here"

Features:

• Auto-detects Gradio API endpoints
• Handles Space URL normalization
• Supports sleeping Space wake-up
• Extracts tool calls if present in response

Example:

# Connect to any HuggingFace Space
evalview connect --adapter hf --endpoint HuggingFaceH4/zephyr-chat

See QUICKSTART_HUGGINGFACE.md for full guide.

---

Streaming Adapter

For JSONL (JSON Lines) streaming APIs.

Use when:

• Your agent streams responses line-by-line
• Each line is a JSON object with event information
• Real-time updates during execution

Configuration:

# .evalview/config.yaml
adapter: streaming  # or 'jsonl' or 'tapescope'
endpoint: http://localhost:3000/api/chat
timeout: 60.0
headers:
  Content-Type: application/json

Expected Event Format:

The streaming adapter recognizes these event types:

{"type": "tool_call", "data": {"name": "search_web", "args": {"query": "..."}}}
{"type": "tool_result", "data": {"result": {"status": "success", "data": [...]}}}
{"type": "token", "data": {"token": "Hello"}}
{"type": "token", "data": {"token": " world"}}
{"type": "final_message", "data": {"text": "Complete response"}}
{"type": "error", "error": "Error message"}

Supported Event Types:

Type	Purpose	Action
tool_call	Tool is being executed	Creates a new step trace
tool_result	Tool finished	Updates last step with result
final_message	Complete response	Sets final output
token	Streaming token (SSE)	Appends to output
error	Error occurred	Captures error message
start, status, thinking	Informational	Logged only

Fallback Behavior:

If no recognized events are found, the adapter:

1. Treats each line as plain text
2. Accumulates all text as final output
3. Still captures timing and basic metrics

---

Vercel AI Adapter

For Vercel AI SDK agents running on HTTP endpoints.

Use when:

• Your agent uses the Vercel AI SDK
• You have an HTTP endpoint serving a Vercel AI agent
• You need to capture streamed responses with tool calls

Configuration:

# .evalview/config.yaml
adapter: vercel-ai
endpoint: http://localhost:3000/api/chat
timeout: 30.0
headers:
  Authorization: Bearer your-api-key

Protocol:

The Vercel AI adapter parses the Vercel AI SDK streaming protocol where each line is prefixed with a type indicator:

0:"Hello "           # Text chunk
2:[{"toolName":"search","args":{"query":"Paris"}}]  # Tool calls array
0:"world"            # More text

Prefix	Type	Content	Action
0:	Text	JSON-encoded string	Appends to output
2:	Tools	JSON array of tool objects	Creates step traces
Other	Metadata	Ignored	No action

Tool Call Format:

Tool calls are provided as a JSON array with toolName and args fields:

[
  {
    "toolName": "search",
    "args": {"query": "Paris"}
  },
  {
    "toolName": "format",
    "args": {"text": "Paris is the capital"}
  }
]

These are automatically mapped to the standard name and arguments fields.

Example Request/Response:

The adapter sends:

{
  "query": "What is the capital of France?",
  "context": {}
}

The agent streams back:

0:"Searching "
2:[{"toolName":"web_search","args":{"query":"capital of France"}}]
0:" for the capital..."

Features:

• Streams tool calls as they arrive
• Accumulates text fragments into final output
• Handles malformed JSON gracefully (logs warning, continues)
• Extracts and records all tool calls in trace steps

Custom Adapters

Build a custom adapter for your specific agent implementation.

Basic Template

# evalview/adapters/my_adapter.py
from datetime import datetime
from typing import Any, Optional, Dict
from evalview.adapters.base import AgentAdapter
from evalview.core.types import (
    ExecutionTrace,
    StepTrace,
    StepMetrics,
    ExecutionMetrics,
)

class MyCustomAdapter(AgentAdapter):
    """Adapter for my custom agent."""

    def __init__(
        self,
        endpoint: str,
        headers: Optional[Dict[str, str]] = None,
        timeout: float = 30.0,
    ):
        self.endpoint = endpoint
        self.headers = headers or {}
        self.timeout = timeout

    @property
    def name(self) -> str:
        return "my-adapter"

    async def execute(
        self,
        query: str,
        context: Optional[Dict[str, Any]] = None
    ) -> ExecutionTrace:
        """Execute agent and capture trace."""
        start_time = datetime.now()

        # TODO: Call your agent API
        response = await self._call_agent(query, context)

        # TODO: Parse response and extract steps
        steps = self._parse_steps(response)

        # TODO: Get final output
        final_output = response.get("output", "")

        end_time = datetime.now()
        total_latency = (end_time - start_time).total_seconds() * 1000

        return ExecutionTrace(
            session_id=response.get("session_id", "custom-session"),
            start_time=start_time,
            end_time=end_time,
            steps=steps,
            final_output=final_output,
            metrics=ExecutionMetrics(
                total_cost=sum(s.metrics.cost for s in steps),
                total_latency=total_latency,
                total_tokens=sum(s.metrics.tokens or 0 for s in steps),
            ),
        )

    async def _call_agent(self, query: str, context: dict) -> dict:
        """Make API call to your agent."""
        import httpx

        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.post(
                self.endpoint,
                json={"query": query, **context},
                headers=self.headers,
            )
            response.raise_for_status()
            return response.json()

    def _parse_steps(self, response: dict) -> list:
        """Convert your response format to StepTrace objects."""
        steps = []

        for step_data in response.get("steps", []):
            step = StepTrace(
                step_id=step_data["id"],
                step_name=step_data["name"],
                tool_name=step_data.get("tool"),
                parameters=step_data.get("parameters", {}),
                output=step_data.get("output"),
                success=step_data.get("success", True),
                error=step_data.get("error"),
                metrics=StepMetrics(
                    latency=step_data.get("latency", 0.0),
                    cost=step_data.get("cost", 0.0),
                    tokens=step_data.get("tokens"),
                ),
            )
            steps.append(step)

        return steps

    async def health_check(self) -> bool:
        """Check if agent endpoint is reachable."""
        try:
            import httpx
            async with httpx.AsyncClient(timeout=5.0) as client:
                response = await client.get(self.endpoint)
                return response.status_code == 200
        except Exception:
            return False

Register Your Adapter

# evalview/cli.py

from evalview.adapters.my_adapter import MyCustomAdapter

# In _run_async function:
adapter_type = config.get("adapter", "http")
if adapter_type == "my-adapter":
    adapter = MyCustomAdapter(
        endpoint=config["endpoint"],
        headers=config.get("headers", {}),
        timeout=config.get("timeout", 30.0),
    )

Adapter Examples

LangServe Streaming

class LangServeAdapter(AgentAdapter):
    """Adapter for LangServe streaming endpoints."""

    async def execute(self, query: str, context=None) -> ExecutionTrace:
        start_time = datetime.now()
        steps = []
        final_output = ""

        async with httpx.AsyncClient(timeout=self.timeout) as client:
            async with client.stream(
                "POST",
                f"{self.endpoint}/stream",
                json={"input": query},
            ) as response:
                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        data = json.loads(line[6:])
                        if "output" in data:
                            final_output = data["output"]

        end_time = datetime.now()
        # ... build ExecutionTrace

OpenAI Responses API

class OpenAIAssistantsAdapter(AgentAdapter):
    """Adapter for the OpenAI Responses API (formerly Assistants)."""

    async def execute(self, query: str, context=None) -> ExecutionTrace:
        import openai

        client = openai.AsyncOpenAI()
        start_time = datetime.now()

        # Conversations replace threads for multi-turn state
        conversation = await client.conversations.create()

        # A response replaces the message + run + polling loop
        response = await client.responses.create(
            model="gpt-4o",
            conversation=conversation.id,
            input=query,
            tools=[{"type": "code_interpreter", "container": {"type": "auto"}}],
        )

        # Extract steps from typed output items (replaces run steps)
        steps = []
        for item in response.output:
            if item.type == "function_call":
                steps.append(StepTrace(
                    step_id=item.id,
                    tool_name=item.name,
                    parameters=json.loads(item.arguments),
                    # ... more fields
                ))
            elif item.type == "code_interpreter_call":
                steps.append(StepTrace(
                    step_id=item.id,
                    tool_name="code_interpreter",
                    parameters={"input": item.code},
                    # ... more fields
                ))

        # Final message text
        final_output = response.output_text

        end_time = datetime.now()
        # ... build ExecutionTrace

CrewAI

class CrewAIAdapter(AgentAdapter):
    """Adapter for CrewAI agents."""

    async def execute(self, query: str, context=None) -> ExecutionTrace:
        # CrewAI typically runs synchronously, wrap in thread
        from concurrent.futures import ThreadPoolExecutor
        import json

        def run_crew():
            # Import your crew
            from my_crew import MyCrew

            crew = MyCrew()
            result = crew.kickoff(inputs={"query": query})
            return result

        start_time = datetime.now()

        with ThreadPoolExecutor() as executor:
            future = executor.submit(run_crew)
            result = future.result(timeout=self.timeout)

        end_time = datetime.now()

        # Parse CrewAI result
        # ... convert to ExecutionTrace

Testing Your Adapter

# test_my_adapter.py
import asyncio
from evalview.adapters.my_adapter import MyCustomAdapter

async def test_adapter():
    adapter = MyCustomAdapter(
        endpoint="http://localhost:3000/api/agent"
    )

    # Test execution
    trace = await adapter.execute("Test query")

    print(f"Session: {trace.session_id}")
    print(f"Steps: {len(trace.steps)}")
    print(f"Output: {trace.final_output}")
    print(f"Latency: {trace.metrics.total_latency}ms")

asyncio.run(test_adapter())

Best Practices

1. Always capture timing - Record start/end times accurately
2. Handle errors gracefully - Catch exceptions and log them
3. Set reasonable timeouts - Don't wait forever for responses
4. Validate responses - Check for required fields before parsing
5. Log verbosely - Use logger for debugging (respects DEBUG env var)
6. Test with real data - Use actual agent responses during development
7. Document event formats - Explain what your adapter expects

Need Help?

• See evalview/adapters/http_adapter.py for simple example
• See evalview/adapters/tapescope_adapter.py for streaming example
• Open an issue on GitHub with your use case
• Check DEBUGGING.md for troubleshooting

Contributing Adapters

Have an adapter for a popular framework? We'd love to include it!

1. Create adapter in evalview/adapters/your_adapter.py
2. Add tests in tests/adapters/test_your_adapter.py
3. Document in this file
4. Submit PR with examples

Popular frameworks we'd love adapters for:

• AutoGPT
• BabyAGI
• SuperAGI
• LangGraph
• Haystack

---

Related Documentation

• Getting Started — Install EvalView and run your first test
• Framework Support — Supported AI agent frameworks and setup
• YAML Schema — Test case schema with adapter and adapter_config fields
• Trace Specification — Execution trace format that all adapters produce
• Backend Requirements — What your agent API needs to expose
• Troubleshooting — Common adapter connection and parsing issues