LlamaIndex Agent Server Adapter

An adapter for hosting LlamaIndex Workflows as Azure AI Agent Server endpoints.

Installation

pip install llamaindex-agentserver-adapter

With OpenAI support:

pip install llamaindex-agentserver-adapter[openai]

With Azure OpenAI support:

pip install llamaindex-agentserver-adapter[azure-openai]

Quick Start

from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent
from llamaindex_agentserver import from_workflow


class MyWorkflow(Workflow):
    @step
    async def process(self, ev: StartEvent) -> StopEvent:
        user_input = ev.input
        return StopEvent(result=f"Processed: {user_input}")


# Create and run the adapter
workflow = MyWorkflow(timeout=60)
adapter = from_workflow(workflow)
adapter.run(port=8088)

Features

• HTTP Endpoints: Exposes your workflow via /runs and /responses endpoints
• Streaming Support: Full support for streaming responses using Server-Sent Events
• Human-in-the-Loop: Support for InputRequiredEvent and HumanResponseEvent patterns
• Tracing: Integration with Azure Application Insights and OpenTelemetry
• Health Checks: Built-in /liveness and /readiness endpoints

API Reference

from_workflow(workflow, state_converter=None, timeout=None)

Creates an adapter for a LlamaIndex Workflow.

Parameters:

• workflow: The LlamaIndex Workflow instance to adapt
• state_converter: Optional custom state converter for non-standard input/output formats
• timeout: Optional timeout override for workflow execution

Returns: A LlamaIndexWorkflowAdapter instance

Request Format

The adapter accepts requests in the OpenAI Responses API format:

{
  "input": "Hello, world!",
  "instructions": "You are a helpful assistant.",
  "stream": false,
  "metadata": {}
}

Or with structured messages:

{
  "input": [
    {"type": "message", "role": "user", "content": "Hello!"},
    {"type": "message", "role": "assistant", "content": "Hi there!"},
    {"type": "message", "role": "user", "content": "How are you?"}
  ],
  "stream": true
}

Testing Your Server

# Non-streaming request
curl -X POST http://localhost:8088/runs \
  -H "Content-Type: application/json" \
  -d '{"input": "Hello, how are you?"}'

# Streaming request
curl -X POST http://localhost:8088/runs \
  -H "Content-Type: application/json" \
  -d '{"input": "Tell me a story", "stream": true}'

# Health check
curl http://localhost:8088/liveness

Custom State Converter

For workflows with custom input/output requirements, implement a custom state converter:

from llamaindex_agentserver.models import LlamaIndexWorkflowStateConverter


class MyStateConverter(LlamaIndexWorkflowStateConverter):
    def supports_streaming(self, context):
        return True

    def request_to_workflow_input(self, context):
        # Convert request to your workflow's expected input format
        return {"custom_input": context.request.get("input")}

    def result_to_response(self, result, context):
        # Convert workflow result to Response object
        ...

    async def stream_to_response_stream(self, event_stream, context):
        # Convert streaming events to response stream events
        ...


adapter = from_workflow(workflow, state_converter=MyStateConverter())

Environment Variables

Examples

See the examples/ directory:

• simple_workflow.py - Basic chat workflow with LLM
• hitl_workflow.py - Human-in-the-loop approval workflow

Requirements

• Python 3.10+
• azure-ai-agentserver-core (Azure AI Agent Server core library)
• llama-index-core and llama-index-workflows

Development

# Clone the repository
git clone https://github.com/yourusername/llamaindex-agentserver-adapter.git
cd llamaindex-agentserver-adapter

# Install in development mode
pip install -e ".[dev]"

# Run tests
pytest

# Format code
ruff format .

# Lint
ruff check .

License

MIT License - See LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.