run-llama/workflows-py
1
0
Fork 0
mirror of https://github.com/run-llama/workflows-py.git synced 2026-07-21 04:05:25 -04:00
Code Issues 18 Packages Projects Releases 10 Wiki Activity

Separate server/client, and move to llama_agents namespace (#277)

Browse Source 
* prep packages

* wrong coverage detection I think

* add ts deps?

* extend path for namespaces

* the big move, import issues though

* llama_index.workflows

* llama_agents

* alias may be working

* add type checker ignores for workflows secondary namespace, update ty, and add re-exports

* Add descriptions to new client and server packages

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Rather, commit the outputs

* keep the py files separate

* ignore examples deps

* Create afraid-books-own.md

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Adrian Lyjak
2026-01-29 18:16:02 -05:00
committed by GitHub
parent 2d5974de0c
commit db90f89b74
70 changed files with 1417 additions and 1066 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/afraid-books-own.md
+7
View File
@@ -0,0 +1,7 @@
---
"llama-index-utils-workflow": patch
"llama-index-workflows-client": patch
"llama-index-workflows-server": patch
---
Separate server/client to their own packages under a llama_agents namespace
.github/workflows/test.yml
+1 -1
View File
@@ -62,7 +62,7 @@ jobs:
- name: Run tests with coverage
if: matrix.python-version == env.COV_PYTHON_VERSION
run: uv run --all-extras --directory ${{ matrix.package_dir }} -- pytest --cov --cov-report=xml
run: uv run --all-extras --directory ${{ matrix.package_dir }} -- pytest --cov=src --cov-report=xml
- name: Report Coveralls
if: matrix.python-version == env.COV_PYTHON_VERSION
.pre-commit-config.yaml
+1 -1
View File
@@ -34,6 +34,7 @@ repos:
rev: v1.15.0
hooks:
- id: mypy
exclude: ^packages/(llama-index-workflows-server|llama-index-workflows-client|llama-index-utils-workflow|workflows-dev|llama-index-integration-tests)/tests/|^packages/llama-index-workflows/src/llama_agents/|^packages/llama-index-workflows/tests/test_llama_agents_alias\.py
additional_dependencies:
[
"types-Deprecated",
@@ -52,7 +53,6 @@ repos:
--ignore-missing-imports,
--python-version=3.11,
]
exclude: ^packages/(llama-index-utils-workflow|llama-index-integration-tests)/tests/
- repo: https://github.com/DetachHead/basedpyright-pre-commit-mirror
rev: 1.31.1
docs/src/content/docs/llamaagents/workflows/deployment.md
+4 -4
View File
@@ -21,7 +21,7 @@ import asyncio
from workflows import Workflow, step
from workflows.context import Context
from workflows.events import Event, StartEvent, StopEvent
from workflows.server import WorkflowServer
from llama_agents.server import WorkflowServer
class StreamEvent(Event):
@@ -290,7 +290,7 @@ In order to interact with a deployed `WorkflowServer` programmatically, beyond t
Assuming you are running the server example from above, we can use `WorkflowClient` in the following way:
```python
from workflows.client import WorkflowClient
from llama_agents.client import WorkflowClient
async def main():
client = WorkflowClient(base_url="http://0.0.0.0:8080")
@@ -330,7 +330,7 @@ from workflows.events import (
InputRequiredEvent,
HumanResponseEvent,
)
from workflows.server import WorkflowServer
from llama_agents.server import WorkflowServer
class RequestEvent(InputRequiredEvent):
prompt: str
@@ -358,7 +358,7 @@ await server.serve("0.0.0.0", "8080")
You can now run the workflow and, when the human interaction is required, send the human response back:
```python
from workflows.client import WorkflowClient
from llama_agents.client import WorkflowClient
client = WorkflowClient(base_url="http://0.0.0.0:8080")
handler = await client.run_workflow_nowait("human")
examples/client/base/workflow_client.py
+1 -1
View File
@@ -1,8 +1,8 @@
import asyncio
from typing import Literal
from llama_agents.client import WorkflowClient
from pydantic import Field
from workflows.client import WorkflowClient
from workflows.events import StartEvent
examples/client/base/workflow_server.py
+1 -1
View File
@@ -1,9 +1,9 @@
from typing import Literal
from llama_agents.server import WorkflowServer
from pydantic import Field
from workflows import Context, Workflow, step
from workflows.events import Event, StartEvent, StopEvent
from workflows.server import WorkflowServer
class InputNumbers(StartEvent):
examples/client/human_in_the_loop/workflow_client_hitl.py
+1 -1
View File
@@ -1,6 +1,6 @@
import asyncio
from workflows.client import WorkflowClient
from llama_agents.client import WorkflowClient
from workflows.events import (
HumanResponseEvent,
StopEvent,
examples/client/human_in_the_loop/workflow_server_hitl.py
+1 -1
View File
@@ -1,3 +1,4 @@
from llama_agents.server import WorkflowServer
from workflows import Workflow, step
from workflows.context import Context
from workflows.events import (
@@ -6,7 +7,6 @@ from workflows.events import (
StartEvent,
StopEvent,
)
from workflows.server import WorkflowServer
class RequestEvent(InputRequiredEvent):
examples/docker/app.py
+1 -1
View File
@@ -1,9 +1,9 @@
import asyncio
from llama_agents.server import WorkflowServer
from workflows import Workflow, step
from workflows.context import Context
from workflows.events import Event, StartEvent, StopEvent
from workflows.server import WorkflowServer
class StreamEvent(Event):
examples/server/fastapi_server_example.ipynb
+218 -311
View File
@@ -1,314 +1,221 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# FastAPI Example"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "H_jVQJE-_mCO"
},
"source": [
"Example demonstrating how to integrate WorkflowServer into an existing FastAPI application.\n",
"\n",
"This example shows how to:\n",
"1. Create a FastAPI application with existing routes\n",
"2. Set up WorkflowServer with workflows\n",
"3. Mount the workflow server as a sub-application\n",
"4. Run the combined application"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"collapsed": true,
"id": "jnPCUjPk_Kfq"
},
"outputs": [],
"source": [
"%pip install llama-index-workflows[server] fastapi"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "wXurfbFgds6s"
},
"source": [
"## Define a workflow and setup the server"
]
},
{
"cell_type": "code",
"execution_count": 1,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "o3oXhBK3_iv8",
"outputId": "0982fbd6-8961-4bb9-a82f-74ba54fb5b27"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Overwriting server.py\n"
]
}
],
"source": [
"%%writefile server.py\n",
"import uvicorn\n",
"from fastapi import FastAPI\n",
"from pydantic import BaseModel\n",
"\n",
"from workflows import Workflow, step\n",
"from workflows.events import StartEvent, StopEvent\n",
"from workflows.server import WorkflowServer\n",
"\n",
"\n",
"class UserModel(BaseModel):\n",
" name: str\n",
" email: str\n",
"\n",
"# Existing FastAPI application with some routes\n",
"app = FastAPI(title=\"My API with Workflows\", version=\"1.0.0\")\n",
"\n",
"\n",
"# Existing API routes\n",
"@app.get(\"/\")\n",
"async def root() -> dict:\n",
" return {\"message\": \"Welcome to My API\"}\n",
"\n",
"\n",
"@app.get(\"/users/{user_id}\")\n",
"async def get_user(user_id: int) -> dict:\n",
" return {\n",
" \"user_id\": user_id,\n",
" \"name\": f\"User {user_id}\",\n",
" \"email\": f\"user{user_id}@example.com\",\n",
" }\n",
"\n",
"\n",
"@app.post(\"/users\")\n",
"async def create_user(user: UserModel) -> dict:\n",
" return {\"message\": f\"Created user {user.name}\", \"user\": user}\n",
"\n",
"\n",
"# Define workflows\n",
"class UserProcessingWorkflow(Workflow):\n",
" @step\n",
" async def process_user(self, ev: StartEvent) -> StopEvent:\n",
" user_data = getattr(ev, \"user_data\", {})\n",
" name = user_data.get(\"name\", \"Unknown\")\n",
" email = user_data.get(\"email\", \"unknown@example.com\")\n",
"\n",
" # Simulate some processing\n",
" processed_data = {\n",
" \"processed_name\": name.upper(),\n",
" \"domain\": email.split(\"@\")[1] if \"@\" in email else \"unknown\",\n",
" \"status\": \"processed\",\n",
" }\n",
"\n",
" return StopEvent(result=processed_data)\n",
"\n",
"\n",
"class NotificationWorkflow(Workflow):\n",
" @step\n",
" async def send_notification(self, ev: StartEvent) -> StopEvent:\n",
" message = getattr(ev, \"message\", \"Default notification\")\n",
" recipient = getattr(ev, \"recipient\", \"admin@example.com\")\n",
"\n",
" # Simulate sending notification\n",
" result = {\n",
" \"notification_id\": \"notif_123\",\n",
" \"message\": message,\n",
" \"recipient\": recipient,\n",
" \"sent_at\": \"2024-01-01T12:00:00Z\",\n",
" \"status\": \"sent\",\n",
" }\n",
"\n",
" return StopEvent(result=result)\n",
"\n",
"\n",
"\n",
"def main() -> None:\n",
" # Create workflow server\n",
" workflow_server = WorkflowServer()\n",
"\n",
" # Register workflows\n",
" workflow_server.add_workflow(\"user_processing\", UserProcessingWorkflow())\n",
" workflow_server.add_workflow(\"notification\", NotificationWorkflow())\n",
"\n",
" # Mount workflow server as sub-application\n",
" app.mount(\"/wf-server\", workflow_server.app)\n",
"\n",
" # run the FastAPI server\n",
" uvicorn.run(app, host=\"0.0.0.0\", port=8000, log_level=\"info\")\n",
"\n",
"\n",
"if __name__ == \"__main__\":\n",
" main()\n"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "V5j--gYldhud"
},
"source": [
"## Run the server in background"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "L5AsbMnk_xuG",
"outputId": "35ba5e76-3e81-43c2-d2cb-504378ed9d41"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"nohup: appending output to 'nohup.out'\n"
]
}
],
"source": [
"!nohup python server.py &"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "bPbCXF_udl4K"
},
"source": [
"## Interact with the server"
]
},
{
"cell_type": "code",
"execution_count": 6,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "H_NVM5ovmyk_",
"outputId": "17738424-95fd-4869-8b0c-19419160ad59"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{\"user_id\":123,\"name\":\"User 123\",\"email\":\"user123@example.com\"}"
]
}
],
"source": [
"# Confirm existing routes are there\n",
"!curl -s http://localhost:8000/users/123"
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "Fm1Z0YGmgmRH",
"outputId": "7150949c-769a-4f6d-d5a8-c2cbb74c599e"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{\"status\":\"healthy\"}"
]
}
],
"source": [
"# Hit the health endpoint to see the workflow server is available at the path /wf-server\n",
"!curl http://localhost:8000/wf-server/health"
]
},
{
"cell_type": "code",
"execution_count": 5,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "TD6tz50jeAkf",
"outputId": "8c2bac46-dc74-4dbd-fe80-2fb4eaac47e5"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{\"workflows\":[\"user_processing\",\"notification\"]}"
]
}
],
"source": [
"# List available workflows\n",
"!curl http://localhost:8000/wf-server/workflows"
]
},
{
"cell_type": "code",
"execution_count": 7,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "wJD4rifveJMt",
"outputId": "c2b124f9-234d-4e10-b257-164e6199c282"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{\"result\":{\"processed_name\":\"ALICE\",\"domain\":\"company.com\",\"status\":\"processed\"}}"
]
}
],
"source": [
"# Run user processing workflow\n",
"!curl -X POST http://localhost:8000/wf-server/workflows/user_processing/run \\\n",
" -H \"Content-Type: application/json\" \\\n",
" -d '{\"kwargs\": {\"user_data\": {\"name\": \"alice\", \"email\": \"alice@company.com\"}}}'"
]
}
],
"metadata": {
"colab": {
"provenance": []
},
"kernelspec": {
"display_name": "Python 3",
"name": "python3"
},
"language_info": {
"name": "python"
}
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# FastAPI Example"
]
},
"nbformat": 4,
"nbformat_minor": 0
{
"cell_type": "markdown",
"metadata": {
"id": "H_jVQJE-_mCO"
},
"source": [
"Example demonstrating how to integrate WorkflowServer into an existing FastAPI application.\n",
"\n",
"This example shows how to:\n",
"1. Create a FastAPI application with existing routes\n",
"2. Set up WorkflowServer with workflows\n",
"3. Mount the workflow server as a sub-application\n",
"4. Run the combined application"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"collapsed": true,
"id": "jnPCUjPk_Kfq"
},
"outputs": [],
"source": [
"%pip install llama-index-workflows-server fastapi"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "V5j--gYldhud"
},
"source": [
"## Run the server in background"
]
},
{
"cell_type": "code",
"execution_count": 8,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "L5AsbMnk_xuG",
"outputId": "35ba5e76-3e81-43c2-d2cb-504378ed9d41"
},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"INFO: Started server process [26386]\n",
"INFO: Waiting for application startup.\n",
"INFO: Application startup complete.\n",
"INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)\n"
]
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"Server started with PID: 26386\n"
]
}
],
"source": [
"import subprocess\n",
"import time\n",
"\n",
"# Start server in background using subprocess\n",
"server_process = subprocess.Popen([\"python\", \"server.py\"])\n",
"time.sleep(2) # Wait for server to start\n",
"print(f\"Server started with PID: {server_process.pid}\")"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "bPbCXF_udl4K"
},
"source": [
"## Interact with the server"
]
},
{
"cell_type": "code",
"execution_count": 9,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "H_NVM5ovmyk_",
"outputId": "17738424-95fd-4869-8b0c-19419160ad59"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"INFO: 127.0.0.1:65035 - \"GET /users/123 HTTP/1.1\" 200 OK\n",
"{\"user_id\":123,\"name\":\"User 123\",\"email\":\"user123@example.com\"}"
]
}
],
"source": [
"# Confirm existing routes are there\n",
"!curl -s http://localhost:8000/users/123"
]
},
{
"cell_type": "code",
"execution_count": 10,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "Fm1Z0YGmgmRH",
"outputId": "7150949c-769a-4f6d-d5a8-c2cbb74c599e"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"INFO: 127.0.0.1:65039 - \"GET /wf-server/health HTTP/1.1\" 200 OK\n",
"{\"status\":\"healthy\",\"loaded_workflows\":0,\"active_workflows\":0,\"idle_workflows\":0}"
]
}
],
"source": [
"# Hit the health endpoint to see the workflow server is available at the path /wf-server\n",
"!curl http://localhost:8000/wf-server/health"
]
},
{
"cell_type": "code",
"execution_count": 11,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "TD6tz50jeAkf",
"outputId": "8c2bac46-dc74-4dbd-fe80-2fb4eaac47e5"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"INFO: 127.0.0.1:65043 - \"GET /wf-server/workflows HTTP/1.1\" 200 OK\n",
"{\"workflows\":[\"user_processing\",\"notification\"]}"
]
}
],
"source": [
"# List available workflows\n",
"!curl http://localhost:8000/wf-server/workflows"
]
},
{
"cell_type": "code",
"execution_count": 12,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "wJD4rifveJMt",
"outputId": "c2b124f9-234d-4e10-b257-164e6199c282"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"INFO: 127.0.0.1:65050 - \"POST /wf-server/workflows/user_processing/run HTTP/1.1\" 200 OK\n",
"{\"handler_id\":\"Mu2qZvQzVu\",\"workflow_name\":\"user_processing\",\"run_id\":\"7hMHiP24AL\",\"error\":null,\"result\":{\"value\":{\"result\":{\"processed_name\":\"ALICE\",\"domain\":\"company.com\",\"status\":\"processed\"}},\"qualified_name\":\"workflows.events.StopEvent\",\"type\":\"StopEvent\",\"types\":null},\"status\":\"completed\",\"started_at\":\"2026-01-29T22:34:33.157083+00:00\",\"updated_at\":\"2026-01-29T22:34:33.157792+00:00\",\"completed_at\":\"2026-01-29T22:34:33.157792+00:00\"}"
]
}
],
"source": [
"# Run user processing workflow\n",
"!curl -X POST http://localhost:8000/wf-server/workflows/user_processing/run \\\n",
" -H \"Content-Type: application/json\" \\\n",
" -d '{\"kwargs\": {\"user_data\": {\"name\": \"alice\", \"email\": \"alice@company.com\"}}}'"
]
}
],
"metadata": {
"colab": {
"provenance": []
},
"kernelspec": {
"display_name": ".venv",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.9.18"
}
},
"nbformat": 4,
"nbformat_minor": 0
}
examples/server/server.py
+90
View File
@@ -0,0 +1,90 @@
import uvicorn
from fastapi import FastAPI
from llama_agents.server import WorkflowServer
from pydantic import BaseModel
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent
class UserModel(BaseModel):
name: str
email: str
# Existing FastAPI application with some routes
app = FastAPI(title="My API with Workflows", version="1.0.0")
# Existing API routes
@app.get("/")
async def root() -> dict:
return {"message": "Welcome to My API"}
@app.get("/users/{user_id}")
async def get_user(user_id: int) -> dict:
return {
"user_id": user_id,
"name": f"User {user_id}",
"email": f"user{user_id}@example.com",
}
@app.post("/users")
async def create_user(user: UserModel) -> dict:
return {"message": f"Created user {user.name}", "user": user}
# Define workflows
class UserProcessingWorkflow(Workflow):
@step
async def process_user(self, ev: StartEvent) -> StopEvent:
user_data = getattr(ev, "user_data", {})
name = user_data.get("name", "Unknown")
email = user_data.get("email", "unknown@example.com")
# Simulate some processing
processed_data = {
"processed_name": name.upper(),
"domain": email.split("@")[1] if "@" in email else "unknown",
"status": "processed",
}
return StopEvent(result=processed_data)
class NotificationWorkflow(Workflow):
@step
async def send_notification(self, ev: StartEvent) -> StopEvent:
message = getattr(ev, "message", "Default notification")
recipient = getattr(ev, "recipient", "admin@example.com")
# Simulate sending notification
result = {
"notification_id": "notif_123",
"message": message,
"recipient": recipient,
"sent_at": "2024-01-01T12:00:00Z",
"status": "sent",
}
return StopEvent(result=result)
def main() -> None:
# Create workflow server
workflow_server = WorkflowServer()
# Register workflows
workflow_server.add_workflow("user_processing", UserProcessingWorkflow())
workflow_server.add_workflow("notification", NotificationWorkflow())
# Mount workflow server as sub-application
app.mount("/wf-server", workflow_server.app)
# run the FastAPI server
uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")
if __name__ == "__main__":
main()
examples/server/server_example.ipynb
+321 -423
View File
@@ -1,426 +1,324 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Workflow Server Example"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "H_jVQJE-_mCO"
},
"source": [
"Example demonstrating how to use the WorkflowServer with event streaming.\n",
"\n",
"This example shows how to:\n",
"1. Create workflows that emit streaming events\n",
"2. Set up the server with event streaming support\n",
"3. Register workflows\n",
"4. Run the server\n",
"5. Make HTTP requests to execute workflows\n",
"6. Stream real-time events from running workflows using the /events endpoint"
]
},
{
"cell_type": "code",
"execution_count": 1,
"metadata": {
"collapsed": true,
"id": "jnPCUjPk_Kfq"
},
"outputs": [],
"source": [
"%pip install llama-index-workflows[server]"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "wXurfbFgds6s"
},
"source": [
"## Define a workflow and setup the server"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "o3oXhBK3_iv8",
"outputId": "c24aa1d6-22c0-44e2-c3ac-2604b8fab9e6"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Writing server.py\n"
]
}
],
"source": [
"%%writefile server.py\n",
"import asyncio\n",
"\n",
"from workflows import Workflow, step\n",
"from workflows.context import Context\n",
"from workflows.events import Event, StartEvent, StopEvent\n",
"from workflows.server import WorkflowServer\n",
"\n",
"\n",
"class StreamEvent(Event):\n",
" sequence: int\n",
"\n",
"\n",
"# Define a simple workflow\n",
"class GreetingWorkflow(Workflow):\n",
" @step\n",
" async def greet(self, ctx: Context, ev: StartEvent) -> StopEvent:\n",
" for i in range(3):\n",
" ctx.write_event_to_stream(StreamEvent(sequence=i))\n",
" await asyncio.sleep(0.3)\n",
"\n",
" name = getattr(ev, \"name\", \"World\")\n",
" return StopEvent(result=f\"Hello, {name}!\")\n",
"\n",
"\n",
"class ProgressEvent(Event):\n",
" step: str\n",
" progress: int\n",
" message: str\n",
"\n",
"\n",
"class MathWorkflow(Workflow):\n",
" @step\n",
" async def calculate(self, ev: StartEvent) -> StopEvent:\n",
" a = getattr(ev, \"a\", 0)\n",
" b = getattr(ev, \"b\", 0)\n",
" operation = getattr(ev, \"operation\", \"add\")\n",
"\n",
" if operation == \"add\":\n",
" result = a + b\n",
" elif operation == \"multiply\":\n",
" result = a * b\n",
" elif operation == \"subtract\":\n",
" result = a - b\n",
" elif operation == \"divide\":\n",
" result = a / b if b != 0 else None\n",
" else:\n",
" result = None\n",
"\n",
" return StopEvent(\n",
" result={\"a\": a, \"b\": b, \"operation\": operation, \"result\": result}\n",
" )\n",
"\n",
"\n",
"class ProcessingWorkflow(Workflow):\n",
" \"\"\"Example workflow that demonstrates event streaming with progress updates.\"\"\"\n",
"\n",
" @step\n",
" async def process(self, ctx: Context, ev: StartEvent) -> StopEvent:\n",
" items = getattr(ev, \"items\", [\"item1\", \"item2\", \"item3\", \"item4\", \"item5\"])\n",
"\n",
" ctx.write_event_to_stream(\n",
" ProgressEvent(\n",
" step=\"start\",\n",
" progress=0,\n",
" message=f\"Starting processing of {len(items)} items\",\n",
" )\n",
" )\n",
"\n",
" results = []\n",
" for i, item in enumerate(items):\n",
" # Simulate processing time\n",
" await asyncio.sleep(0.5)\n",
"\n",
" # Emit progress event\n",
" progress = int((i + 1) / len(items) * 100)\n",
" ctx.write_event_to_stream(\n",
" ProgressEvent(\n",
" step=\"processing\",\n",
" progress=progress,\n",
" message=f\"Processed {item} ({i + 1}/{len(items)})\",\n",
" )\n",
" )\n",
"\n",
" results.append(f\"processed_{item}\")\n",
"\n",
" ctx.write_event_to_stream(\n",
" ProgressEvent(\n",
" step=\"complete\",\n",
" progress=100,\n",
" message=\"Processing completed successfully\",\n",
" )\n",
" )\n",
"\n",
" return StopEvent(result={\"processed_items\": results, \"total\": len(results)})\n",
"\n",
"\n",
"async def main():\n",
" server = WorkflowServer()\n",
"\n",
" # Register workflows\n",
" server.add_workflow(\"greeting\", GreetingWorkflow())\n",
" server.add_workflow(\"math\", MathWorkflow())\n",
" server.add_workflow(\"processing\", ProcessingWorkflow())\n",
"\n",
" await server.serve(host=\"0.0.0.0\", port=8000)\n",
"\n",
"\n",
"if __name__ == \"__main__\":\n",
" asyncio.run(main())"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "V5j--gYldhud"
},
"source": [
"## Run the server in background"
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "L5AsbMnk_xuG",
"outputId": "0cb5ec84-575e-4484-90ab-94ff73e241d8"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"nohup: appending output to 'nohup.out'\n"
]
}
],
"source": [
"!nohup python server.py &"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "bPbCXF_udl4K"
},
"source": [
"## Interact with the server"
]
},
{
"cell_type": "code",
"execution_count": 4,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "Fm1Z0YGmgmRH",
"outputId": "f4c87973-1034-4fe3-c5f4-82d8c0d78db2"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{\"status\":\"healthy\"}"
]
}
],
"source": [
"# Hit the health endpoint to see the server is up and running\n",
"!curl http://localhost:8000/health"
]
},
{
"cell_type": "code",
"execution_count": 5,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "TD6tz50jeAkf",
"outputId": "2a622940-7e0b-445a-9292-e89420026ef6"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{\"workflows\":[\"greeting\",\"math\",\"processing\"]}"
]
}
],
"source": [
"# List available workflows\n",
"!curl http://localhost:8000/workflows"
]
},
{
"cell_type": "code",
"execution_count": 6,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "wJD4rifveJMt",
"outputId": "ab382fc4-8e94-4779-826f-0db4c4986f69"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{\"result\":\"Hello, Alice!\"}"
]
}
],
"source": [
"# Run greeting workflow\n",
"!curl -X POST http://localhost:8000/workflows/greeting/run \\\n",
" -H \"Content-Type: application/json\" \\\n",
" -d '{\"kwargs\": {\"name\": \"Alice\"}}'"
]
},
{
"cell_type": "code",
"execution_count": 7,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "UkO28ZgxeOyT",
"outputId": "b0543b29-e102-4ba9-a44c-4115c89c0e44"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{\"result\":{\"a\":10,\"b\":5,\"operation\":\"multiply\",\"result\":50}}"
]
}
],
"source": [
"# Run math workflow\n",
"!curl -X POST http://localhost:8000/workflows/math/run \\\n",
" -H \"Content-Type: application/json\" \\\n",
" -d '{\"kwargs\": {\"a\": 10, \"b\": 5, \"operation\": \"multiply\"}}'"
]
},
{
"cell_type": "code",
"execution_count": 36,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "-PQ6vt56eTwp",
"outputId": "7c7f0007-ecd5-4c68-8275-0cbecc15c4d8"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Got handler id: vuyyRzWikr\n",
"{\"result\":{\"a\":100,\"b\":25,\"operation\":\"divide\",\"result\":4.0}}"
]
}
],
"source": [
"%%bash\n",
"# Run workflow with nowait\n",
"handler_id=$(curl -sX POST http://localhost:8000/workflows/math/run-nowait \\\n",
" -H \"Content-Type: application/json\" \\\n",
" -d '{\"kwargs\": {\"a\": 100, \"b\": 25, \"operation\": \"divide\"}}' | jq -r \".handler_id\" )\n",
"printf \"Got handler id: ${handler_id}\\n\\n\"\n",
"\n",
"# Wait for the workflow to run in background\n",
"sleep 1\n",
"\n",
"# Fetch the result asynchronously\n",
"curl -s http://localhost:8000/results/${handler_id}"
]
},
{
"cell_type": "code",
"execution_count": 40,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "iR4-fBOujIXd",
"outputId": "b35d418f-bbd6-47b9-b0de-deac1ac26bd1"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Got handler id: nPBSVhI8DO\n",
"\n",
"Streaming events...\n",
"event: __main__.StreamEvent\n",
"data: {\"sequence\": 0}\n",
"event: __main__.StreamEvent\n",
"data: {\"sequence\": 1}\n",
"event: __main__.StreamEvent\n",
"data: {\"sequence\": 2}\n",
"event: workflows.events.StopEvent\n",
"data: {}\n",
"\n",
"Final result:\n",
"{\"result\":\"Hello, Async User!\"}"
]
}
],
"source": [
"%%bash\n",
"# Stream events from workflow\n",
"\n",
"# 1. Run workflow with nowait\n",
"handler_id=$(curl -sX POST http://localhost:8000/workflows/greeting/run-nowait \\\n",
" -H \"Content-Type: application/json\" \\\n",
" -d '{\"kwargs\": {\"name\": \"Async User\"}}' | jq -r \".handler_id\" )\n",
"printf \"Got handler id: ${handler_id}\\n\\n\"\n",
"\n",
"# Wait for the workflow to run in background\n",
"sleep 1\n",
"\n",
"printf \"Streaming events...\\n\"\n",
"# 2. Stream events using Server-Sent Events using SSE format\n",
"curl -s http://localhost:8000/events/$handler_id?sse=true\n",
"\n",
"\n",
"printf \"\\nFinal result:\\n\"\n",
"# 3. Get the final result after events complete\n",
"curl -s http://localhost:8000/results/$handler_id"
]
}
],
"metadata": {
"colab": {
"provenance": []
},
"kernelspec": {
"display_name": "Python 3",
"name": "python3"
},
"language_info": {
"name": "python"
}
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Workflow Server Example"
]
},
"nbformat": 4,
"nbformat_minor": 0
{
"cell_type": "markdown",
"metadata": {
"id": "H_jVQJE-_mCO"
},
"source": [
"Example demonstrating how to use the WorkflowServer with event streaming.\n",
"\n",
"This example shows how to:\n",
"1. Create workflows that emit streaming events\n",
"2. Set up the server with event streaming support\n",
"3. Register workflows\n",
"4. Run the server\n",
"5. Make HTTP requests to execute workflows\n",
"6. Stream real-time events from running workflows using the /events endpoint"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"collapsed": true,
"id": "jnPCUjPk_Kfq"
},
"outputs": [],
"source": [
"%pip install llama-index-workflows-server"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "V5j--gYldhud"
},
"source": [
"## Run the server in background"
]
},
{
"cell_type": "code",
"execution_count": 4,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "L5AsbMnk_xuG",
"outputId": "0cb5ec84-575e-4484-90ab-94ff73e241d8"
},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"INFO: Started server process [27939]\n",
"INFO: Waiting for application startup.\n",
"INFO: Application startup complete.\n",
"INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)\n"
]
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"Server started with PID: 27939\n"
]
}
],
"source": [
"import subprocess\n",
"import time\n",
"\n",
"# Start server in background using subprocess\n",
"server_process = subprocess.Popen([\"python\", \"server.py\"])\n",
"time.sleep(2) # Wait for server to start\n",
"print(f\"Server started with PID: {server_process.pid}\")"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "bPbCXF_udl4K"
},
"source": [
"## Interact with the server"
]
},
{
"cell_type": "code",
"execution_count": 5,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "Fm1Z0YGmgmRH",
"outputId": "f4c87973-1034-4fe3-c5f4-82d8c0d78db2"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"INFO: 127.0.0.1:65182 - \"GET /health HTTP/1.1\" 200 OK\n",
"{\"status\":\"healthy\",\"loaded_workflows\":0,\"active_workflows\":0,\"idle_workflows\":0}"
]
}
],
"source": [
"# Hit the health endpoint to see the server is up and running\n",
"!curl http://localhost:8000/health"
]
},
{
"cell_type": "code",
"execution_count": 6,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "TD6tz50jeAkf",
"outputId": "2a622940-7e0b-445a-9292-e89420026ef6"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"INFO: 127.0.0.1:65184 - \"GET /workflows HTTP/1.1\" 200 OK\n",
"{\"workflows\":[\"greeting\",\"math\",\"processing\"]}"
]
}
],
"source": [
"# List available workflows\n",
"!curl http://localhost:8000/workflows"
]
},
{
"cell_type": "code",
"execution_count": 7,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "wJD4rifveJMt",
"outputId": "ab382fc4-8e94-4779-826f-0db4c4986f69"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"INFO: 127.0.0.1:65190 - \"POST /workflows/greeting/run HTTP/1.1\" 200 OK\n",
"{\"handler_id\":\"3podAAguhA\",\"workflow_name\":\"greeting\",\"run_id\":\"KD7Y2S0e0h\",\"error\":null,\"result\":{\"value\":{\"result\":\"Hello, Alice!\"},\"qualified_name\":\"workflows.events.StopEvent\",\"type\":\"StopEvent\",\"types\":null},\"status\":\"completed\",\"started_at\":\"2026-01-29T22:36:48.596733+00:00\",\"updated_at\":\"2026-01-29T22:36:49.500898+00:00\",\"completed_at\":\"2026-01-29T22:36:49.500898+00:00\"}"
]
}
],
"source": [
"# Run greeting workflow\n",
"!curl -X POST http://localhost:8000/workflows/greeting/run \\\n",
" -H \"Content-Type: application/json\" \\\n",
" -d '{\"kwargs\": {\"name\": \"Alice\"}}'"
]
},
{
"cell_type": "code",
"execution_count": 8,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "UkO28ZgxeOyT",
"outputId": "b0543b29-e102-4ba9-a44c-4115c89c0e44"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"INFO: 127.0.0.1:65196 - \"POST /workflows/math/run HTTP/1.1\" 200 OK\n",
"{\"handler_id\":\"kITlArjbWm\",\"workflow_name\":\"math\",\"run_id\":\"t3jXb1nGtV\",\"error\":null,\"result\":{\"value\":{\"result\":{\"a\":10,\"b\":5,\"operation\":\"multiply\",\"result\":50}},\"qualified_name\":\"workflows.events.StopEvent\",\"type\":\"StopEvent\",\"types\":null},\"status\":\"completed\",\"started_at\":\"2026-01-29T22:36:56.341122+00:00\",\"updated_at\":\"2026-01-29T22:36:56.341812+00:00\",\"completed_at\":\"2026-01-29T22:36:56.341812+00:00\"}"
]
}
],
"source": [
"# Run math workflow\n",
"!curl -X POST http://localhost:8000/workflows/math/run \\\n",
" -H \"Content-Type: application/json\" \\\n",
" -d '{\"kwargs\": {\"a\": 10, \"b\": 5, \"operation\": \"multiply\"}}'"
]
},
{
"cell_type": "code",
"execution_count": 36,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "-PQ6vt56eTwp",
"outputId": "7c7f0007-ecd5-4c68-8275-0cbecc15c4d8"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Got handler id: vuyyRzWikr\n",
"{\"result\":{\"a\":100,\"b\":25,\"operation\":\"divide\",\"result\":4.0}}"
]
}
],
"source": [
"%%bash\n",
"# Run workflow with nowait\n",
"handler_id=$(curl -sX POST http://localhost:8000/workflows/math/run-nowait \\\n",
" -H \"Content-Type: application/json\" \\\n",
" -d '{\"kwargs\": {\"a\": 100, \"b\": 25, \"operation\": \"divide\"}}' | jq -r \".handler_id\" )\n",
"printf \"Got handler id: ${handler_id}\\n\\n\"\n",
"\n",
"# Wait for the workflow to run in background\n",
"sleep 1\n",
"\n",
"# Fetch the result asynchronously\n",
"curl -s http://localhost:8000/results/${handler_id}"
]
},
{
"cell_type": "code",
"execution_count": 9,
"metadata": {
"colab": {
"base_uri": "https://localhost:8080/"
},
"id": "iR4-fBOujIXd",
"outputId": "b35d418f-bbd6-47b9-b0de-deac1ac26bd1"
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"INFO: 127.0.0.1:65202 - \"POST /workflows/greeting/run-nowait HTTP/1.1\" 200 OK\n",
"Got handler id: gbqSi1fzGJ\n",
"\n"
]
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"Streaming events...\n",
"INFO: 127.0.0.1:65204 - \"GET /events/gbqSi1fzGJ?sse=true HTTP/1.1\" 200 OK\n",
"data: {\"value\":{\"sequence\":0},\"qualified_name\":\"__main__.StreamEvent\",\"type\":\"StreamEvent\",\"types\":null}\n",
"\n",
"data: {\"value\":{\"sequence\":1},\"qualified_name\":\"__main__.StreamEvent\",\"type\":\"StreamEvent\",\"types\":null}\n",
"\n",
"data: {\"value\":{\"sequence\":2},\"qualified_name\":\"__main__.StreamEvent\",\"type\":\"StreamEvent\",\"types\":null}\n",
"\n",
"data: {\"value\":{\"result\":\"Hello, Async User!\"},\"qualified_name\":\"workflows.events.StopEvent\",\"type\":\"StopEvent\",\"types\":null}\n",
"\n",
"\n",
"Final result:\n",
"INFO: 127.0.0.1:65206 - \"GET /results/gbqSi1fzGJ HTTP/1.1\" 200 OK\n",
"{\"handler_id\":\"gbqSi1fzGJ\",\"workflow_name\":\"greeting\",\"run_id\":\"M5AtcoxBty\",\"error\":null,\"result\":\"Hello, Async User!\",\"status\":\"completed\",\"started_at\":\"2026-01-29T22:37:02.375925+00:00\",\"updated_at\":\"2026-01-29T22:37:03.279696+00:00\",\"completed_at\":\"2026-01-29T22:37:03.279696+00:00\"}"
]
}
],
"source": [
"%%bash\n",
"# Stream events from workflow\n",
"\n",
"# 1. Run workflow with nowait\n",
"handler_id=$(curl -sX POST http://localhost:8000/workflows/greeting/run-nowait \\\n",
" -H \"Content-Type: application/json\" \\\n",
" -d '{\"kwargs\": {\"name\": \"Async User\"}}' | jq -r \".handler_id\" )\n",
"printf \"Got handler id: ${handler_id}\\n\\n\"\n",
"\n",
"# Wait for the workflow to run in background\n",
"sleep 1\n",
"\n",
"printf \"Streaming events...\\n\"\n",
"# 2. Stream events using Server-Sent Events using SSE format\n",
"curl -s http://localhost:8000/events/$handler_id?sse=true\n",
"\n",
"\n",
"printf \"\\nFinal result:\\n\"\n",
"# 3. Get the final result after events complete\n",
"curl -s http://localhost:8000/results/$handler_id"
]
}
],
"metadata": {
"colab": {
"provenance": []
},
"kernelspec": {
"display_name": ".venv",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.9.18"
}
},
"nbformat": 4,
"nbformat_minor": 0
}
examples/server/server_example.py
+1 -1
View File
@@ -1,5 +1,6 @@
import asyncio
from llama_agents.server import WorkflowServer
from workflows import Workflow, step
from workflows.context import Context
from workflows.events import (
@@ -9,7 +10,6 @@ from workflows.events import (
StartEvent,
StopEvent,
)
from workflows.server import WorkflowServer
class StreamEvent(Event):
packages/llama-index-utils-workflow/package.json
+4 -1
View File
@@ -3,5 +3,8 @@
"version": "0.8.0",
"private": false,
"license": "MIT",
"scripts": {}
"scripts": {},
"dependencies": {
"llama-index-workflows": "workspace:*"
}
}
packages/llama-index-workflows/src/workflows/server/sqlite/migrations/__init__.py → packages/llama-index-workflows-client/README.md
View File
packages/llama-index-workflows-client/package.json
+10
View File
@@ -0,0 +1,10 @@
{
"name": "llama-index-workflows-client",
"version": "0.1.0",
"private": false,
"license": "MIT",
"scripts": {},
"dependencies": {
"llama-index-workflows": "workspace:*"
}
}
packages/llama-index-workflows-client/pyproject.toml
+20
View File
@@ -0,0 +1,20 @@
[build-system]
requires = ["uv_build>=0.9.6,<0.10.0"]
build-backend = "uv_build"
[project]
name = "llama-index-workflows-client"
version = "0.1.0"
description = "HTTP client for connecting to and interacting with LlamaIndex workflow servers"
readme = "README.md"
requires-python = ">=3.9"
dependencies = [
"httpx>=0.28.1,<1",
"llama-index-workflows>=2.12.0,<3.0.0"
]
[tool.uv.build-backend]
module-name = "llama_agents.client"
[tool.uv.sources]
llama-index-workflows = {workspace = true}
packages/llama-index-workflows-client/src/llama_agents/client/__init__.py
+3
View File
@@ -0,0 +1,3 @@
from .client import WorkflowClient
__all__ = ["WorkflowClient"]
packages/llama-index-workflows/src/workflows/client/client.py → packages/llama-index-workflows-client/src/llama_agents/client/client.py
+3 -3
View File
@@ -12,10 +12,10 @@ from typing import (
)
import httpx
from workflows import Context
from workflows.events import Event, StartEvent
from workflows.protocol import (
from .protocol import (
CancelHandlerResponse,
HandlerData,
HandlersListResponse,
@@ -24,7 +24,7 @@ from workflows.protocol import (
Status,
WorkflowsListResponse,
)
from workflows.protocol.serializable_events import (
from .protocol.serializable_events import (
EventEnvelope,
EventEnvelopeWithMetadata,
)
packages/llama-index-workflows-client/src/llama_agents/client/protocol/__init__.py
+84
View File
@@ -0,0 +1,84 @@
from __future__ import annotations
from typing import Any, Literal
from pydantic import BaseModel, Field
from workflows.representation import WorkflowGraph
from .serializable_events import EventEnvelopeWithMetadata
# Shared protocol types between client and server
# Mirrors server.store Status
Status = Literal["running", "completed", "failed", "cancelled"]
def is_status_completed(status: Status) -> bool:
return status in {"completed", "failed", "cancelled"}
class HandlerData(BaseModel):
handler_id: str
workflow_name: str
run_id: str | None
error: str | None
result: EventEnvelopeWithMetadata | None
status: Status
started_at: str
updated_at: str | None
completed_at: str | None
class HandlersListResponse(BaseModel):
handlers: list[HandlerData]
class HealthResponse(BaseModel):
status: Literal["healthy"]
loaded_workflows: int = Field(
description="Number of workflow handlers currently loaded in memory"
)
active_workflows: int = Field(
description="Number of workflow handlers that are active (not idle)"
)
idle_workflows: int = Field(description="Number of workflow handlers that are idle")
class WorkflowsListResponse(BaseModel):
workflows: list[str]
class SendEventResponse(BaseModel):
status: Literal["sent"]
class CancelHandlerResponse(BaseModel):
status: Literal["deleted", "cancelled"]
class WorkflowSchemaResponse(BaseModel):
start: dict[str, Any]
stop: dict[str, Any]
class WorkflowEventsListResponse(BaseModel):
events: list[dict[str, Any]]
class WorkflowGraphResponse(BaseModel):
graph: WorkflowGraph
__all__ = [
"Status",
"is_status_completed",
"HandlerData",
"HandlersListResponse",
"HealthResponse",
"WorkflowsListResponse",
"SendEventResponse",
"CancelHandlerResponse",
"WorkflowSchemaResponse",
"WorkflowEventsListResponse",
"WorkflowGraphResponse",
]
packages/llama-index-workflows/src/workflows/protocol/serializable_events.py → packages/llama-index-workflows-client/src/llama_agents/client/protocol/serializable_events.py
-1
View File
@@ -7,7 +7,6 @@ import json
from typing import Any, Type
from pydantic import BaseModel, ValidationError, model_validator
from workflows.context.utils import import_module_from_qualified_name
from workflows.events import Event
packages/llama-index-workflows/tests/client/client_workflows.py → packages/llama-index-workflows-client/tests/client/client_test_workflows.py
+3
View File
@@ -1,3 +1,6 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
import random
from workflows import Context, Workflow, step
packages/llama-index-workflows/tests/client/test_client.py → packages/llama-index-workflows-client/tests/client/test_client.py
+8 -7
View File
@@ -1,18 +1,19 @@
import httpx
import pytest
from httpx import ASGITransport, AsyncClient
from workflows.client import WorkflowClient
from workflows.protocol.serializable_events import EventEnvelopeWithMetadata
from workflows.server.memory_workflow_store import MemoryWorkflowStore
from workflows.server.server import WorkflowServer
from .client_workflows import ( # type: ignore[import]
from client_test_workflows import (
GreetEvent,
InputEvent,
OutputEvent,
crashing_wf,
greeting_wf,
)
from httpx import ASGITransport, AsyncClient
from llama_agents.client import WorkflowClient
from llama_agents.client.protocol.serializable_events import (
EventEnvelopeWithMetadata,
)
from llama_agents.server.memory_workflow_store import MemoryWorkflowStore
from llama_agents.server.server import WorkflowServer
@pytest.fixture()
packages/llama-index-workflows/tests/client/__init__.py → packages/llama-index-workflows-client/tests/protocol/__init__.py
View File
packages/llama-index-workflows/tests/protocol/test_serializable_events.py → packages/llama-index-workflows-client/tests/protocol/test_serializable_events.py
+5 -5
View File
@@ -4,17 +4,17 @@
import json
import pytest
from llama_agents.client.protocol.serializable_events import (
EventEnvelope,
EventEnvelopeWithMetadata,
EventValidationError,
)
from workflows.events import (
Event,
StepState,
StepStateChanged,
StopEvent,
)
from workflows.protocol.serializable_events import (
EventEnvelope,
EventEnvelopeWithMetadata,
EventValidationError,
)
def test_envelope_user_defined_event() -> None:
packages/llama-index-workflows/tests/server/__init__.py → packages/llama-index-workflows-server/README.md
View File
packages/llama-index-workflows-server/package.json
+11
View File
@@ -0,0 +1,11 @@
{
"name": "llama-index-workflows-server",
"version": "0.1.0",
"private": false,
"license": "MIT",
"scripts": {},
"dependencies": {
"llama-index-workflows": "workspace:*",
"llama-index-workflows-client": "workspace:*"
}
}
packages/llama-index-workflows-server/pyproject.toml
+26
View File
@@ -0,0 +1,26 @@
[build-system]
requires = ["uv_build>=0.9.6,<0.10.0"]
build-backend = "uv_build"
[project]
name = "llama-index-workflows-server"
version = "0.1.0"
description = "HTTP server for deploying and serving LlamaIndex workflows as web services"
readme = "README.md"
requires-python = ">=3.9"
dependencies = [
"llama-index-workflows>=2.12.0,<3.0.0",
"llama-index-workflows-client>=0.1.0,<0.2.0",
"starlette>=0.39.0",
"uvicorn>=0.32.0"
]
[project.scripts]
llama-index-workflows-server = "llama_agents.server.__main__:run_server"
[tool.uv.build-backend]
module-name = "llama_agents.server"
[tool.uv.sources]
llama-index-workflows = {workspace = true}
llama-index-workflows-client = {workspace = true}
packages/llama-index-workflows-server/src/llama_agents/server/__init__.py
+18
View File
@@ -0,0 +1,18 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
from .abstract_workflow_store import (
AbstractWorkflowStore,
HandlerQuery,
PersistentHandler,
)
from .server import WorkflowServer
from .sqlite.sqlite_workflow_store import SqliteWorkflowStore
__all__ = [
"WorkflowServer",
"AbstractWorkflowStore",
"HandlerQuery",
"PersistentHandler",
"SqliteWorkflowStore",
]
packages/llama-index-workflows/src/workflows/server/__main__.py → packages/llama-index-workflows-server/src/llama_agents/server/__main__.py
+1 -1
View File
@@ -6,7 +6,7 @@ from pathlib import Path
import uvicorn
from workflows.server.server import WorkflowServer
from .server import WorkflowServer
def run_server() -> None:
packages/llama-index-workflows/src/workflows/server/abstract_workflow_store.py → packages/llama-index-workflows-server/src/llama_agents/server/abstract_workflow_store.py
-1
View File
@@ -10,7 +10,6 @@ from pydantic import (
field_serializer,
field_validator,
)
from workflows.context import JsonSerializer
from workflows.events import StopEvent
packages/llama-index-workflows/src/workflows/server/keyed_lock.py → packages/llama-index-workflows-server/src/llama_agents/server/keyed_lock.py
View File
packages/llama-index-workflows/src/workflows/server/memory_workflow_store.py → packages/llama-index-workflows-server/src/llama_agents/server/memory_workflow_store.py
+1 -1
View File
@@ -1,6 +1,6 @@
from typing import Dict, List
from workflows.server.abstract_workflow_store import (
from .abstract_workflow_store import (
AbstractWorkflowStore,
HandlerQuery,
PersistentHandler,
packages/llama-index-workflows/src/workflows/server/server.py → packages/llama-index-workflows-server/src/llama_agents/server/server.py
+23 -23
View File
@@ -13,6 +13,22 @@ from pathlib import Path
from typing import Any, AsyncGenerator, Awaitable, Callable, cast
import uvicorn
from llama_agents.client.protocol import (
CancelHandlerResponse,
HandlerData,
HandlersListResponse,
HealthResponse,
SendEventResponse,
WorkflowEventsListResponse,
WorkflowGraphResponse,
WorkflowSchemaResponse,
is_status_completed,
)
from llama_agents.client.protocol.serializable_events import (
EventEnvelope,
EventEnvelopeWithMetadata,
EventValidationError,
)
from llama_index_instrumentation.dispatcher import instrument_tags
from starlette.applications import Starlette
from starlette.exceptions import HTTPException
@@ -23,7 +39,6 @@ from starlette.responses import JSONResponse, StreamingResponse
from starlette.routing import Route
from starlette.schemas import SchemaGenerator
from starlette.staticfiles import StaticFiles
from workflows import Context, Workflow
from workflows.events import (
Event,
@@ -36,34 +51,19 @@ from workflows.events import (
WorkflowIdleEvent,
)
from workflows.handler import WorkflowHandler
from workflows.protocol import (
CancelHandlerResponse,
HandlerData,
HandlersListResponse,
HealthResponse,
SendEventResponse,
WorkflowEventsListResponse,
WorkflowGraphResponse,
WorkflowSchemaResponse,
is_status_completed,
)
from workflows.protocol.serializable_events import (
EventEnvelope,
EventEnvelopeWithMetadata,
EventValidationError,
)
from workflows.representation import get_workflow_representation
from workflows.server.abstract_workflow_store import (
# Protocol models are used on the client side; server responds with plain dicts
from workflows.utils import _nanoid as nanoid
from .abstract_workflow_store import (
AbstractWorkflowStore,
HandlerQuery,
PersistentHandler,
Status,
)
from workflows.server.keyed_lock import KeyedLock
from workflows.server.memory_workflow_store import MemoryWorkflowStore
# Protocol models are used on the client side; server responds with plain dicts
from workflows.utils import _nanoid as nanoid
from .keyed_lock import KeyedLock
from .memory_workflow_store import MemoryWorkflowStore
logger = logging.getLogger()
packages/llama-index-workflows-server/src/llama_agents/server/sqlite/__init__.py
View File
packages/llama-index-workflows/src/workflows/server/sqlite/migrate.py → packages/llama-index-workflows-server/src/llama_agents/server/sqlite/migrate.py
+1 -1
View File
@@ -12,7 +12,7 @@ from importlib import import_module, resources
logger = logging.getLogger(__name__)
_MIGRATIONS_PKG = "workflows.server.sqlite.migrations"
_MIGRATIONS_PKG = "llama_agents.server.sqlite.migrations"
_USER_VERSION_PATTERN = re.compile(r"pragma\s+user_version\s*=\s*(\d+)", re.IGNORECASE)
packages/llama-index-workflows/src/workflows/server/sqlite/migrations/0001_init.sql → packages/llama-index-workflows-server/src/llama_agents/server/sqlite/migrations/0001_init.sql
View File
packages/llama-index-workflows/src/workflows/server/sqlite/migrations/0002_extend_handlers.sql → packages/llama-index-workflows-server/src/llama_agents/server/sqlite/migrations/0002_extend_handlers.sql
View File
packages/llama-index-workflows/src/workflows/server/sqlite/migrations/0003_add_idle_since.sql → packages/llama-index-workflows-server/src/llama_agents/server/sqlite/migrations/0003_add_idle_since.sql
View File
packages/llama-index-workflows-server/src/llama_agents/server/sqlite/migrations/__init__.py
View File
packages/llama-index-workflows/src/workflows/server/sqlite/sqlite_workflow_store.py → packages/llama-index-workflows-server/src/llama_agents/server/sqlite/sqlite_workflow_store.py
+3 -2
View File
@@ -4,12 +4,13 @@ from datetime import datetime
from typing import List, Optional, Sequence, Tuple
from workflows.context import JsonSerializer
from workflows.server.abstract_workflow_store import (
from ..abstract_workflow_store import (
AbstractWorkflowStore,
HandlerQuery,
PersistentHandler,
)
from workflows.server.sqlite.migrate import run_migrations
from .migrate import run_migrations
class SqliteWorkflowStore(AbstractWorkflowStore):
packages/llama-index-workflows/src/workflows/server/static/index.html → packages/llama-index-workflows-server/src/llama_agents/server/static/index.html
View File
packages/llama-index-workflows-server/tests/server/conftest.py
+2
View File
@@ -0,0 +1,2 @@
# Re-export fixtures from server_test_fixtures for pytest discovery
from server_test_fixtures import * # noqa: F401, F403
packages/llama-index-workflows/tests/server/conftest.py → packages/llama-index-workflows-server/tests/server/server_test_fixtures.py
+93
View File
@@ -3,8 +3,15 @@
import asyncio
import socket
import time
from contextlib import asynccontextmanager
from typing import AsyncGenerator, Awaitable, Callable, TypeVar
import httpx
import pytest
import uvicorn
from llama_agents.server import WorkflowServer
from workflows import Context, Workflow, step
from workflows.events import (
Event,
@@ -14,6 +21,92 @@ from workflows.events import (
StopEvent,
)
T = TypeVar("T")
async def async_yield(iterations: int = 10) -> None:
"""Yield to the event loop multiple times to let async tasks run."""
for _ in range(iterations):
await asyncio.sleep(0)
async def wait_for_passing(
func: Callable[[], Awaitable[T]],
max_duration: float = 5.0,
interval: float = 0.05,
) -> T:
start_time = time.monotonic()
last_exception = None
while time.monotonic() - start_time < max_duration:
remaining_duration = max_duration - (time.monotonic() - start_time)
try:
return await asyncio.wait_for(func(), timeout=remaining_duration)
except Exception as e:
last_exception = e
await asyncio.sleep(interval)
if last_exception:
raise last_exception
else:
func_name = getattr(func, "__name__", repr(func))
raise TimeoutError(
f"Function {func_name} timed out after {max_duration} seconds"
)
@asynccontextmanager
async def live_server(
server_factory: Callable[[], WorkflowServer],
) -> AsyncGenerator[tuple[str, WorkflowServer], None]:
"""Start a live HTTP server for testing with atomic port acquisition."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
sock.bind(("127.0.0.1", 0))
sock.listen(128)
port = sock.getsockname()[1]
server = server_factory()
config = uvicorn.Config(
server.app,
host="127.0.0.1",
port=port,
log_level="error",
loop="asyncio",
)
uv_server = uvicorn.Server(config)
task = asyncio.create_task(uv_server.serve(sockets=[sock]))
base_url = f"http://127.0.0.1:{port}"
async with httpx.AsyncClient(base_url=base_url, timeout=1.0) as client:
for _ in range(50):
try:
resp = await client.get("/health")
if resp.status_code == 200:
break
except Exception:
pass
await asyncio.sleep(0.01)
else:
uv_server.should_exit = True
await task
raise RuntimeError("Live server did not start in time")
try:
yield base_url, server
finally:
uv_server.should_exit = True
try:
await task
finally:
await server.stop()
finally:
try:
sock.close()
except Exception:
pass
class SimpleTestWorkflow(Workflow):
@step
packages/llama-index-workflows/tests/server/test_handler_serialization.py → packages/llama-index-workflows-server/tests/server/test_handler_serialization.py
+3 -3
View File
@@ -8,12 +8,12 @@ from datetime import datetime, timezone
from unittest.mock import MagicMock
import pytest
from llama_agents.client.protocol import HandlerData
from llama_agents.server.memory_workflow_store import MemoryWorkflowStore
from llama_agents.server.server import _WorkflowHandler
from workflows.context import Context
from workflows.events import Event, StopEvent
from workflows.handler import WorkflowHandler
from workflows.protocol import HandlerData
from workflows.server.memory_workflow_store import MemoryWorkflowStore
from workflows.server.server import _WorkflowHandler
class MyStopEvent(StopEvent):
packages/llama-index-workflows/tests/server/test_idle_release.py → packages/llama-index-workflows-server/tests/server/test_idle_release.py
+6 -7
View File
@@ -9,17 +9,16 @@ from typing import Any, Optional
import pytest
import time_machine
from httpx import ASGITransport, AsyncClient
from workflows import Context, Workflow, step
from workflows.events import HumanResponseEvent, StartEvent, StopEvent
from workflows.server.abstract_workflow_store import (
from llama_agents.server.abstract_workflow_store import (
HandlerQuery,
PersistentHandler,
Status,
)
from workflows.server.memory_workflow_store import MemoryWorkflowStore
from workflows.server.server import WorkflowServer, _WorkflowHandler
from tests.server.util import async_yield, wait_for_passing
from llama_agents.server.memory_workflow_store import MemoryWorkflowStore
from llama_agents.server.server import WorkflowServer, _WorkflowHandler
from server_test_fixtures import async_yield, wait_for_passing
from workflows import Context, Workflow, step
from workflows.events import HumanResponseEvent, StartEvent, StopEvent
class WaitableExternalEvent(HumanResponseEvent):
packages/llama-index-workflows/tests/server/test_idle_release_live_http.py → packages/llama-index-workflows-server/tests/server/test_idle_release_live_http.py
+6 -7
View File
@@ -7,16 +7,15 @@ from __future__ import annotations
from datetime import timedelta
import pytest
from workflows import Context, Workflow, step
from workflows.client.client import WorkflowClient
from workflows.events import Event, StartEvent, StopEvent, WorkflowIdleEvent
from workflows.server import WorkflowServer
from workflows.server.memory_workflow_store import MemoryWorkflowStore
from .util import (
from llama_agents.client.client import WorkflowClient
from llama_agents.server import WorkflowServer
from llama_agents.server.memory_workflow_store import MemoryWorkflowStore
from server_test_fixtures import (
live_server, # type: ignore[import]
wait_for_passing, # type: ignore[import]
)
from workflows import Context, Workflow, step
from workflows.events import Event, StartEvent, StopEvent, WorkflowIdleEvent
class WaitableExternalEvent(Event):
packages/llama-index-workflows/tests/server/test_keyed_lock.py → packages/llama-index-workflows-server/tests/server/test_keyed_lock.py
+1 -1
View File
@@ -7,7 +7,7 @@ from __future__ import annotations
import asyncio
import pytest
from workflows.server.keyed_lock import KeyedLock
from llama_agents.server.keyed_lock import KeyedLock
@pytest.fixture
packages/llama-index-workflows/tests/server/test_main.py → packages/llama-index-workflows-server/tests/server/test_main.py
+6 -6
View File
@@ -7,7 +7,7 @@ from typing import Any
from unittest.mock import patch
import pytest
from workflows.server.__main__ import run_server
from llama_agents.server.__main__ import run_server
def test_no_file_path_argument(capsys: Any) -> None:
@@ -78,7 +78,7 @@ def test_workflow_server_with_custom_name(tmp_path: Path) -> None:
# Create a test Python file with WorkflowServer instance named differently
test_file = tmp_path / "custom_server.py"
test_file.write_text("""
from workflows.server.server import WorkflowServer
from llama_agents.server.server import WorkflowServer
# WorkflowServer with custom name
my_app = WorkflowServer()
@@ -100,7 +100,7 @@ def test_multiple_workflow_servers_uses_first(tmp_path: Path) -> None:
"""Test that when multiple WorkflowServer instances exist, the first one is used."""
test_file = tmp_path / "multiple_servers.py"
test_file.write_text("""
from workflows.server.server import WorkflowServer
from llama_agents.server.server import WorkflowServer
# Multiple WorkflowServer instances
first_server = WorkflowServer()
@@ -119,7 +119,7 @@ def test_environment_variables(tmp_path: Path) -> None:
"""Test that environment variables are used for host and port."""
test_file = tmp_path / "env_test.py"
test_file.write_text("""
from workflows.server.server import WorkflowServer
from llama_agents.server.server import WorkflowServer
server = WorkflowServer()
""")
@@ -149,7 +149,7 @@ def test_module_loading_error(capsys: Any, tmp_path: Path) -> None:
# Create a file with syntax error
test_file = tmp_path / "syntax_error.py"
test_file.write_text("""
from workflows.server.server import WorkflowServer
from llama_agents.server.server import WorkflowServer
# Syntax error
def invalid_syntax(
@@ -186,7 +186,7 @@ def test_non_workflow_server_objects_ignored(tmp_path: Path) -> None:
"""Test that objects that aren't WorkflowServer instances are ignored."""
test_file = tmp_path / "mixed_objects.py"
test_file.write_text("""
from workflows.server.server import WorkflowServer
from llama_agents.server.server import WorkflowServer
# Various non-WorkflowServer objects
string_var = "not a server"
packages/llama-index-workflows/tests/server/test_memory_workflow_store.py → packages/llama-index-workflows-server/tests/server/test_memory_workflow_store.py
+5 -2
View File
@@ -1,9 +1,12 @@
from datetime import datetime, timezone
import pytest
from llama_agents.server.abstract_workflow_store import (
HandlerQuery,
PersistentHandler,
)
from llama_agents.server.memory_workflow_store import MemoryWorkflowStore
from workflows.events import StopEvent
from workflows.server.abstract_workflow_store import HandlerQuery, PersistentHandler
from workflows.server.memory_workflow_store import MemoryWorkflowStore
@pytest.mark.asyncio
packages/llama-index-workflows/tests/server/test_migrations.py → packages/llama-index-workflows-server/tests/server/test_migrations.py
+1 -1
View File
@@ -4,7 +4,7 @@ import sqlite3
from pathlib import Path
from typing import List
from workflows.server.sqlite.migrate import run_migrations
from llama_agents.server.sqlite.migrate import run_migrations
def _get_table_columns(conn: sqlite3.Connection, table: str) -> List[str]:
packages/llama-index-workflows/tests/server/test_openapi_schema.py → packages/llama-index-workflows-server/tests/server/test_openapi_schema.py
+1 -1
View File
@@ -1,8 +1,8 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
from llama_agents.server import WorkflowServer
from workflows import Workflow
from workflows.server import WorkflowServer
def test_openapi_schema_includes_all_routes(simple_test_workflow: Workflow) -> None:
packages/llama-index-workflows/tests/server/test_persistent_handler_serialization.py → packages/llama-index-workflows-server/tests/server/test_persistent_handler_serialization.py
+1 -1
View File
@@ -2,8 +2,8 @@ from __future__ import annotations
from typing import Any, cast
from llama_agents.server.abstract_workflow_store import PersistentHandler
from workflows.events import StopEvent
from workflows.server.abstract_workflow_store import PersistentHandler
def _base_handler_kwargs() -> dict[str, Any]:
packages/llama-index-workflows/tests/server/test_server.py → packages/llama-index-workflows-server/tests/server/test_server.py
+5 -5
View File
@@ -5,8 +5,8 @@ from typing import Any
from unittest.mock import AsyncMock, Mock, patch
import pytest
from llama_agents.server import WorkflowServer
from starlette.middleware import Middleware
from workflows.server import WorkflowServer
from workflows.workflow import Workflow
@@ -31,8 +31,8 @@ def test_add_workflow(simple_test_workflow: Workflow) -> None:
@pytest.mark.asyncio
@patch("workflows.server.server.uvicorn.Server")
@patch("workflows.server.server.uvicorn.Config")
@patch("llama_agents.server.server.uvicorn.Server")
@patch("llama_agents.server.server.uvicorn.Config")
async def test_serve(mock_config: Any, mock_server: Any) -> None:
server = WorkflowServer()
mock_server_instance = AsyncMock()
@@ -45,8 +45,8 @@ async def test_serve(mock_config: Any, mock_server: Any) -> None:
@pytest.mark.asyncio
@patch("workflows.server.server.uvicorn.Server")
@patch("workflows.server.server.uvicorn.Config")
@patch("llama_agents.server.server.uvicorn.Server")
@patch("llama_agents.server.server.uvicorn.Config")
async def test_serve_with_uvicorn_config(mock_config: Any, mock_server: Any) -> None:
server = WorkflowServer()
mock_server_instance = AsyncMock()
packages/llama-index-workflows/tests/server/test_server_endpoints.py → packages/llama-index-workflows-server/tests/server/test_server_endpoints.py
+12 -8
View File
@@ -14,20 +14,24 @@ from typing import Any, AsyncGenerator, AsyncIterator
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient, Response
from llama_agents.server import WorkflowServer
from llama_agents.server.abstract_workflow_store import (
HandlerQuery,
PersistentHandler,
)
from llama_agents.server.memory_workflow_store import MemoryWorkflowStore
from llama_index_instrumentation.dispatcher import active_instrument_tags
from server_test_fixtures import (
ExternalEvent, # type: ignore[import]
wait_for_passing, # type: ignore[import]
)
from workflows import Context, step
# Prepare the event to send
from workflows.context.serializers import JsonSerializer
from workflows.events import StartEvent, StopEvent
from workflows.server import WorkflowServer
from workflows.server.abstract_workflow_store import HandlerQuery, PersistentHandler
from workflows.server.memory_workflow_store import MemoryWorkflowStore
from workflows.workflow import Workflow
from .conftest import ExternalEvent # type: ignore[import]
from .util import wait_for_passing # type: ignore[import]
class CustomStopEvent(StopEvent):
message: str
@@ -431,7 +435,7 @@ async def test_stream_events_success(client: AsyncClient) -> None:
events.append(event_data)
stream_events = [
e for e in events if e["qualified_name"] == "tests.server.conftest.StreamEvent"
e for e in events if e["qualified_name"] == "server_test_fixtures.StreamEvent"
]
assert len(stream_events) == 3
for i, event in enumerate(stream_events):
@@ -479,7 +483,7 @@ async def test_stream_events_sse(client: AsyncClient) -> None:
stream_events = [
e
for e in events
if e["data"]["qualified_name"] == "tests.server.conftest.StreamEvent"
if e["data"]["qualified_name"] == "server_test_fixtures.StreamEvent"
]
assert len(stream_events) == 2
packages/llama-index-workflows/tests/server/test_server_live_http.py → packages/llama-index-workflows-server/tests/server/test_server_live_http.py
+7 -8
View File
@@ -8,17 +8,16 @@ import contextlib
from typing import AsyncGenerator
import pytest
from workflows import Workflow
from workflows.client.client import WorkflowClient
from workflows.events import StopEvent
from workflows.server import WorkflowServer
from .conftest import ( # type: ignore[import]
from llama_agents.client.client import WorkflowClient
from llama_agents.server import WorkflowServer
from server_test_fixtures import ( # type: ignore[import]
ExternalEvent,
RequestedExternalEvent,
wait_for_passing, # type: ignore[import]
)
from .util import live_server as live_server_ctx # type: ignore[import]
from .util import wait_for_passing # type: ignore[import]
from server_test_fixtures import live_server as live_server_ctx # type: ignore[import]
from workflows import Workflow
from workflows.events import StopEvent
@pytest.fixture
packages/llama-index-workflows/tests/server/test_server_persistence.py → packages/llama-index-workflows-server/tests/server/test_server_persistence.py
+11 -9
View File
@@ -5,18 +5,20 @@ from typing import AsyncGenerator
import pytest
from httpx import ASGITransport, AsyncClient
from workflows import Context
from workflows.events import Event, InternalDispatchEvent, StopEvent
from workflows.server import WorkflowServer
from workflows.server.abstract_workflow_store import HandlerQuery, PersistentHandler
from workflows.server.memory_workflow_store import MemoryWorkflowStore
from workflows.workflow import Workflow
from .conftest import ( # type: ignore[import]
from llama_agents.server import WorkflowServer
from llama_agents.server.abstract_workflow_store import (
HandlerQuery,
PersistentHandler,
)
from llama_agents.server.memory_workflow_store import MemoryWorkflowStore
from server_test_fixtures import ( # type: ignore[import]
ExternalEvent,
RequestedExternalEvent,
wait_for_passing, # type: ignore[import]
)
from .util import wait_for_passing # type: ignore[import]
from workflows import Context
from workflows.events import Event, InternalDispatchEvent, StopEvent
from workflows.workflow import Workflow
@pytest.fixture
packages/llama-index-workflows/tests/server/test_sqlite_workflow_store.py → packages/llama-index-workflows-server/tests/server/test_sqlite_workflow_store.py
+7 -2
View File
@@ -1,9 +1,14 @@
from pathlib import Path
import pytest
from llama_agents.server.abstract_workflow_store import (
HandlerQuery,
PersistentHandler,
)
from llama_agents.server.sqlite.sqlite_workflow_store import (
SqliteWorkflowStore,
)
from workflows.events import StopEvent
from workflows.server.abstract_workflow_store import HandlerQuery, PersistentHandler
from workflows.server.sqlite.sqlite_workflow_store import SqliteWorkflowStore
@pytest.mark.asyncio
packages/llama-index-workflows/pyproject.toml
+7 -3
View File
@@ -34,9 +34,12 @@ dependencies = [
"typing-extensions>=4.6.0"
]
# backward compatibility, for when these were embedded.
# Optionals are now just an alias to additionally install
# a less pinned version of the client/server packages.
[project.optional-dependencies]
server = ["starlette>=0.39.0", "uvicorn>=0.32.0"]
client = ["httpx>=0.28.1,<1"]
server = ["llama-index-workflows-server>=0.1.0,<1.0.0"]
client = ["llama-index-workflows-client>=0.1.0,<1.0.0"]
[tool.hatch.envs.server]
features = ["server"]
@@ -50,4 +53,5 @@ asyncio_mode = "auto"
testpaths = ["tests"]
[tool.uv.build-backend]
module-name = "workflows"
module-name = ["workflows", "llama_agents.workflows"]
namespace = true
packages/llama-index-workflows/src/llama_agents/workflows/__init__.py
+68
View File
@@ -0,0 +1,68 @@
# SPDX-License-Identifier: MIT
# Alias: llama_agents.workflows -> workflows
#
# This module makes the entire `workflows` package available under
# `llama_agents.workflows`, including all sub-modules. It uses a
# custom meta-path finder to lazily redirect any import of
# `llama_agents.workflows.<sub>` to `workflows.<sub>`.
from __future__ import annotations
import importlib
import sys
from importlib.abc import Loader, MetaPathFinder
from importlib.machinery import ModuleSpec
from types import ModuleType
from typing import Sequence
_ALIAS_PREFIX = "llama_agents.workflows"
_REAL_PREFIX = "workflows"
class _AliasLoader(Loader):
"""Loader that returns an already-imported module from sys.modules."""
def __init__(self, real_name: str) -> None:
self.real_name = real_name
def create_module(self, spec: ModuleSpec) -> ModuleType | None:
return importlib.import_module(self.real_name)
def exec_module(self, module: ModuleType) -> None:
# Module is already fully initialized by the real import.
pass
class _AliasFinder(MetaPathFinder):
"""Meta-path finder that redirects llama_agents.workflows.* to workflows.*"""
def find_module(
self, fullname: str, path: Sequence[str] | None = None
) -> Loader | None:
# Python 3.9 compat - some tools call find_module instead of find_spec
spec = self.find_spec(fullname, path)
if spec and spec.loader:
return spec.loader # type: ignore[return-value]
return None
def find_spec(
self,
fullname: str,
path: Sequence[str] | None = None,
target: ModuleType | None = None,
) -> ModuleSpec | None:
# Only handle llama_agents.workflows.* (not the root itself)
if not fullname.startswith(_ALIAS_PREFIX + "."):
return None
suffix = fullname[len(_ALIAS_PREFIX) :]
real_name = _REAL_PREFIX + suffix
return ModuleSpec(fullname, _AliasLoader(real_name))
# Install the finder once
if not any(isinstance(f, _AliasFinder) for f in sys.meta_path):
sys.meta_path.append(_AliasFinder())
# Re-export everything from the real workflows package
from workflows import * # noqa: E402, F403
from workflows import __all__ # noqa: E402, F401
packages/llama-index-workflows/src/workflows/__init__.py
+5
View File
@@ -1,10 +1,15 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
from pkgutil import extend_path
from .context import Context
from .decorators import step
from .workflow import Workflow
__path__ = extend_path(__path__, __name__)
__all__ = [
"Context",
"Workflow",
packages/llama-index-workflows/src/workflows/client/__init__.py
+21 -1
View File
@@ -1,3 +1,23 @@
from .client import WorkflowClient
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
"""Re-export client components from the optional llama-index-workflows-client package."""
import warnings
warnings.warn(
"Importing from 'workflows.client' is deprecated. "
"Install 'llama-index-workflows-client' and use "
"'from llama_agents.client import ...' instead.",
DeprecationWarning,
stacklevel=2,
)
try:
from llama_agents.client import WorkflowClient
except ImportError as e:
raise ImportError(
"workflows.client requires the 'client' extra. "
"Install with: pip install 'llama-index-workflows[client]'"
) from e
__all__ = ["WorkflowClient"]
packages/llama-index-workflows/src/workflows/protocol/__init__.py
+29 -66
View File
@@ -1,73 +1,36 @@
from __future__ import annotations
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
"""Re-export protocol types from the optional llama-index-workflows-client package."""
from typing import Any, Literal
import warnings
from pydantic import BaseModel, Field
warnings.warn(
"Importing from 'workflows.protocol' is deprecated. "
"Install 'llama-index-workflows-client' and use "
"'from llama_agents.client.protocol import ...' instead.",
DeprecationWarning,
stacklevel=2,
)
from workflows.protocol.serializable_events import EventEnvelopeWithMetadata
from workflows.representation import WorkflowGraph
# Shared protocol types between client and server
# Mirrors server.store Status
Status = Literal["running", "completed", "failed", "cancelled"]
def is_status_completed(status: Status) -> bool:
return status in {"completed", "failed", "cancelled"}
class HandlerData(BaseModel):
handler_id: str
workflow_name: str
run_id: str | None
error: str | None
result: EventEnvelopeWithMetadata | None
status: Status
started_at: str
updated_at: str | None
completed_at: str | None
class HandlersListResponse(BaseModel):
handlers: list[HandlerData]
class HealthResponse(BaseModel):
status: Literal["healthy"]
loaded_workflows: int = Field(
description="Number of workflow handlers currently loaded in memory"
try:
from llama_agents.client.protocol import (
CancelHandlerResponse,
HandlerData,
HandlersListResponse,
HealthResponse,
SendEventResponse,
Status,
WorkflowEventsListResponse,
WorkflowGraphResponse,
WorkflowSchemaResponse,
WorkflowsListResponse,
is_status_completed,
)
active_workflows: int = Field(
description="Number of workflow handlers that are active (not idle)"
)
idle_workflows: int = Field(description="Number of workflow handlers that are idle")
class WorkflowsListResponse(BaseModel):
workflows: list[str]
class SendEventResponse(BaseModel):
status: Literal["sent"]
class CancelHandlerResponse(BaseModel):
status: Literal["deleted", "cancelled"]
class WorkflowSchemaResponse(BaseModel):
start: dict[str, Any]
stop: dict[str, Any]
class WorkflowEventsListResponse(BaseModel):
events: list[dict[str, Any]]
class WorkflowGraphResponse(BaseModel):
graph: WorkflowGraph
except ImportError as e:
raise ImportError(
"workflows.protocol requires the 'client' extra. "
"Install with: pip install 'llama-index-workflows[client]'"
) from e
__all__ = [
"Status",
packages/llama-index-workflows/src/workflows/server/__init__.py
+23 -6
View File
@@ -1,13 +1,30 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
"""Re-export server components from the optional llama-index-workflows-server package."""
from .abstract_workflow_store import (
AbstractWorkflowStore,
HandlerQuery,
PersistentHandler,
import warnings
warnings.warn(
"Importing from 'workflows.server' is deprecated. "
"Install 'llama-index-workflows-server' and use "
"'from llama_agents.server import ...' instead.",
DeprecationWarning,
stacklevel=2,
)
from .server import WorkflowServer
from .sqlite.sqlite_workflow_store import SqliteWorkflowStore
try:
from llama_agents.server import (
AbstractWorkflowStore,
HandlerQuery,
PersistentHandler,
SqliteWorkflowStore,
WorkflowServer,
)
except ImportError as e:
raise ImportError(
"workflows.server requires the 'server' extra. "
"Install with: pip install 'llama-index-workflows[server]'"
) from e
__all__ = [
"WorkflowServer",
packages/llama-index-workflows/src/workflows/server/sqlite/__init__.py
+13
View File
@@ -0,0 +1,13 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
"""Re-export sqlite components from the optional llama-index-workflows-server package."""
try:
from llama_agents.server.sqlite.sqlite_workflow_store import SqliteWorkflowStore
except ImportError as e:
raise ImportError(
"workflows.server.sqlite requires the 'server' extra. "
"Install with: pip install 'llama-index-workflows[server]'"
) from e
__all__ = ["SqliteWorkflowStore"]
packages/llama-index-workflows/tests/server/util.py
-127
View File
@@ -1,127 +0,0 @@
import asyncio
import socket
import time
from contextlib import asynccontextmanager
from typing import AsyncGenerator, Awaitable, Callable, TypeVar
import httpx
import uvicorn
from workflows.server import WorkflowServer
T = TypeVar("T")
async def async_yield(iterations: int = 10) -> None:
"""Yield to the event loop multiple times to let async tasks run.
Use this when you need to let async tasks make progress without
waiting for real time to pass. Useful with time_machine when time
is frozen (tick=False).
"""
for _ in range(iterations):
await asyncio.sleep(0)
async def wait_for_passing(
func: Callable[[], Awaitable[T]],
max_duration: float = 5.0,
interval: float = 0.05,
) -> T:
start_time = time.monotonic()
last_exception = None
while time.monotonic() - start_time < max_duration:
remaining_duration = max_duration - (time.monotonic() - start_time)
try:
return await asyncio.wait_for(func(), timeout=remaining_duration)
except Exception as e:
last_exception = e
await asyncio.sleep(interval)
if last_exception:
raise last_exception
else:
func_name = getattr(func, "__name__", repr(func))
raise TimeoutError(
f"Function {func_name} timed out after {max_duration} seconds"
)
@asynccontextmanager
async def live_server(
server_factory: Callable[[], WorkflowServer],
) -> AsyncGenerator[tuple[str, WorkflowServer], None]:
"""Start a live HTTP server for testing with atomic port acquisition.
This context manager handles:
- Atomic port acquisition (no race condition with parallel tests)
- Server startup with health check
- Graceful shutdown
Args:
server_factory: A callable that creates and configures a WorkflowServer.
This allows tests to customize workflows, idle_release_timeout, etc.
Yields:
A tuple of (base_url, server) for making requests and inspecting state.
Example:
def make_server() -> WorkflowServer:
server = WorkflowServer(idle_release_timeout=timedelta(seconds=1))
server.add_workflow("test", MyWorkflow())
return server
async with live_server(make_server) as (base_url, server):
client = WorkflowClient(base_url=base_url)
# ... run tests
"""
# Create socket and bind atomically - prevents race condition in parallel tests
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
sock.bind(("127.0.0.1", 0))
sock.listen(128)
port = sock.getsockname()[1]
server = server_factory()
config = uvicorn.Config(
server.app,
host="127.0.0.1",
port=port,
log_level="error",
loop="asyncio",
)
uv_server = uvicorn.Server(config)
# Start server in background task with our pre-bound socket
task = asyncio.create_task(uv_server.serve(sockets=[sock]))
# Wait until server responds on /health or timeout
base_url = f"http://127.0.0.1:{port}"
async with httpx.AsyncClient(base_url=base_url, timeout=1.0) as client:
for _ in range(50): # ~0.5s max wait
try:
resp = await client.get("/health")
if resp.status_code == 200:
break
except Exception:
pass
await asyncio.sleep(0.01)
else:
uv_server.should_exit = True
await task
raise RuntimeError("Live server did not start in time")
try:
yield base_url, server
finally:
uv_server.should_exit = True
try:
await task
finally:
await server.stop()
finally:
# Socket is managed by uvicorn after serve() starts, but close if we fail early
try:
sock.close()
except Exception:
pass
packages/llama-index-workflows/tests/test_llama_agents_alias.py
+123
View File
@@ -0,0 +1,123 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 LlamaIndex Inc.
# pyright: reportMissingImports=false
"""Tests that the llama_agents.workflows alias resolves correctly."""
from __future__ import annotations
import pytest
def test_top_level_import() -> None:
from llama_agents.workflows import Context, Workflow, step
assert Workflow is not None
assert Context is not None
assert callable(step)
def test_top_level_identity() -> None:
import llama_agents.workflows as alias
import workflows
assert alias.Workflow is workflows.Workflow
assert alias.Context is workflows.Context
assert alias.step is workflows.step
def test_events_submodule() -> None:
from llama_agents.workflows.events import (
Event,
StartEvent,
StopEvent,
)
assert Event is not None
assert issubclass(StartEvent, Event)
assert issubclass(StopEvent, Event)
def test_events_submodule_identity() -> None:
import llama_agents.workflows.events as alias_events
import workflows.events
assert alias_events is workflows.events
assert alias_events.Event is workflows.events.Event
def test_context_submodule() -> None:
from llama_agents.workflows.context import Context
assert Context is not None
def test_context_submodule_identity() -> None:
import llama_agents.workflows.context as alias_ctx
import workflows.context
assert alias_ctx is workflows.context
def test_workflow_submodule() -> None:
from llama_agents.workflows.workflow import Workflow
assert Workflow is not None
def test_decorators_submodule() -> None:
from llama_agents.workflows.decorators import step
assert callable(step)
def test_errors_submodule() -> None:
from llama_agents.workflows.errors import (
WorkflowRuntimeError,
WorkflowTimeoutError,
WorkflowValidationError,
)
assert issubclass(WorkflowRuntimeError, Exception)
assert issubclass(WorkflowTimeoutError, Exception)
assert issubclass(WorkflowValidationError, Exception)
def test_handler_submodule() -> None:
from llama_agents.workflows.handler import WorkflowHandler
assert WorkflowHandler is not None
def test_testing_submodule() -> None:
from llama_agents.workflows.testing import WorkflowTestRunner
assert WorkflowTestRunner is not None
def test_deep_submodule() -> None:
from llama_agents.workflows.context.context import Context
assert Context is not None
def test_dunder_all_reexported() -> None:
import llama_agents.workflows as alias
import workflows
assert alias.__all__ == workflows.__all__
@pytest.mark.asyncio
async def test_alias_workflow_runs() -> None:
"""A workflow defined via the alias module actually executes."""
from llama_agents.workflows import Context, Workflow, step
from llama_agents.workflows.events import StartEvent, StopEvent
class HelloWorkflow(Workflow):
@step
async def say_hello(self, ctx: Context, ev: StartEvent) -> StopEvent:
return StopEvent(result="hello")
wf = HelloWorkflow()
result = await wf.run()
assert result == "hello"
pnpm-lock.yaml
Generated
+21 -1
View File
@@ -87,10 +87,29 @@ importers:
specifier: ^7.0.0
version: 7.0.0(typedoc@0.28.15(typescript@5.9.3))
packages/llama-index-utils-workflow: {}
packages/llama-index-utils-workflow:
dependencies:
llama-index-workflows:
specifier: workspace:*
version: link:../llama-index-workflows
packages/llama-index-workflows: {}
packages/llama-index-workflows-client:
dependencies:
llama-index-workflows:
specifier: workspace:*
version: link:../llama-index-workflows
packages/llama-index-workflows-server:
dependencies:
llama-index-workflows:
specifier: workspace:*
version: link:../llama-index-workflows
llama-index-workflows-client:
specifier: workspace:*
version: link:../llama-index-workflows-client
packages:
'@antfu/install-pkg@1.1.0':
@@ -3453,6 +3472,7 @@ packages:
whatwg-encoding@3.1.1:
resolution: {integrity: sha512-6qN4hJdMwfYBtE3YBTTHhoeuUrDBPZmbQaxWAqSALV/MeEnR5z1xd8UKud2RAkFoPkmB+hli1TZSnyi84xz1vQ==}
engines: {node: '>=18'}
deprecated: Use @exodus/bytes instead for a more spec-conformant and faster implementation
whatwg-mimetype@4.0.0:
resolution: {integrity: sha512-QaKxh0eNIi2mE9p2vEdzfagOKHCcj1pJ56EEHGQOVxp8r9/iszLUUV7v89x9O1p/T+NlTM5W7jW6+cz4Fq1YVg==}
pyproject.toml
+28 -3
View File
@@ -24,6 +24,14 @@ typeCheckingMode = "standard"
root = "packages/llama-index-workflows"
pythonVersion = "3.9"
[[tool.basedpyright.executionEnvironments]]
root = "packages/llama-index-workflows-client"
pythonVersion = "3.9"
[[tool.basedpyright.executionEnvironments]]
root = "packages/llama-index-workflows-server"
pythonVersion = "3.9"
[[tool.basedpyright.executionEnvironments]]
root = "packages/llama-index-utils-workflow"
pythonVersion = "3.9"
@@ -39,10 +47,22 @@ pythonVersion = "3.13"
[[tool.basedpyright.executionEnvironments]]
root = "examples"
pythonVersion = "3.14"
reportMissingImports = false
[tool.coverage.run]
omit = ["**/tests/*"]
[tool.mypy]
explicit_package_bases = true
exclude = [
"packages/llama-index-workflows-server/tests/",
"packages/llama-index-workflows-client/tests/",
"packages/llama-index-utils-workflow/tests/",
"packages/workflows-dev/tests/",
"packages/llama-index-workflows/src/llama_agents/",
"packages/llama-index-workflows/tests/test_llama_agents_alias.py"
]
[tool.pytest.ini_options]
asyncio_mode = "auto"
@@ -75,19 +95,24 @@ select = [
"__main__.py" = ["T201"]
[tool.ty.src]
exclude = ["**/uv.lock", "**/.gitignore"]
# Exclude llama_agents.workflows alias (dynamic import aliasing not understood by static analyzers)
exclude = ["**/uv.lock", "**/.gitignore", "**/llama_agents/workflows/**", "**/test_llama_agents_alias.py"]
[tool.uv.sources]
llama-index-workflows = {workspace = true}
llama-index-utils-workflow = {workspace = true}
workflows-dev = {workspace = true}
llama-index-integration-tests = {workspace = true}
llama-index-workflows-client = {workspace = true}
llama-index-workflows-server = {workspace = true}
workflows-dev = {workspace = true}
[tool.uv.workspace]
members = [
"docs/api_docs",
"packages/llama-index-workflows",
"packages/llama-index-utils-workflow",
"packages/workflows-dev",
"packages/llama-index-integration-tests",
"docs/api_docs"
"packages/llama-index-workflows-client",
"packages/llama-index-workflows-server"
]
uv.lock
Generated
+40 -6
View File
@@ -14,6 +14,8 @@ members = [
"llama-index-integration-tests",
"llama-index-utils-workflow",
"llama-index-workflows",
"llama-index-workflows-client",
"llama-index-workflows-server",
"workflows-dev",
]
@@ -1720,11 +1722,10 @@ dependencies = [
[package.optional-dependencies]
client = [
{ name = "httpx" },
{ name = "llama-index-workflows-client" },
]
server = [
{ name = "starlette" },
{ name = "uvicorn" },
{ name = "llama-index-workflows-server" },
]
[package.dev-dependencies]
@@ -1748,12 +1749,11 @@ dev = [
[package.metadata]
requires-dist = [
{ name = "eval-type-backport", marker = "python_full_version < '3.10'", specifier = ">=0.2.2" },
{ name = "httpx", marker = "extra == 'client'", specifier = ">=0.28.1,<1" },
{ name = "llama-index-instrumentation", specifier = ">=0.1.0" },
{ name = "llama-index-workflows-client", marker = "extra == 'client'", editable = "packages/llama-index-workflows-client" },
{ name = "llama-index-workflows-server", marker = "extra == 'server'", editable = "packages/llama-index-workflows-server" },
{ name = "pydantic", specifier = ">=2.11.5" },
{ name = "starlette", marker = "extra == 'server'", specifier = ">=0.39.0" },
{ name = "typing-extensions", specifier = ">=4.6.0" },
{ name = "uvicorn", marker = "extra == 'server'", specifier = ">=0.32.0" },
]
provides-extras = ["server", "client"]
@@ -1775,6 +1775,40 @@ dev = [
{ name = "uvicorn", specifier = ">=0.32.0" },
]
[[package]]
name = "llama-index-workflows-client"
version = "0.1.0"
source = { editable = "packages/llama-index-workflows-client" }
dependencies = [
{ name = "httpx" },
{ name = "llama-index-workflows" },
]
[package.metadata]
requires-dist = [
{ name = "httpx", specifier = ">=0.28.1,<1" },
{ name = "llama-index-workflows", editable = "packages/llama-index-workflows" },
]
[[package]]
name = "llama-index-workflows-server"
version = "0.1.0"
source = { editable = "packages/llama-index-workflows-server" }
dependencies = [
{ name = "llama-index-workflows" },
{ name = "llama-index-workflows-client" },
{ name = "starlette" },
{ name = "uvicorn" },
]
[package.metadata]
requires-dist = [
{ name = "llama-index-workflows", editable = "packages/llama-index-workflows" },
{ name = "llama-index-workflows-client", editable = "packages/llama-index-workflows-client" },
{ name = "starlette", specifier = ">=0.39.0" },
{ name = "uvicorn", specifier = ">=0.32.0" },
]
[[package]]
name = "markdown"
version = "3.9"