GraphQL API Reference¶

This document provides a complete reference for the apflow GraphQL API, built on Strawberry GraphQL.

Overview¶

The GraphQL API provides: - Typed schema with full introspection support - Queries for reading tasks and task trees - Mutations for creating, updating, cancelling, and deleting tasks - Subscriptions for real-time task status updates via WebSocket - GraphiQL interactive playground for exploring the schema

Installation¶

Install apflow with the GraphQL extra:

pip install apflow[graphql]

Configuration¶

Set the API protocol to graphql:

export APFLOW_API_PROTOCOL=graphql

Start the server:

apflow serve
# Or: python -m apflow.api.main

The GraphQL endpoint is available at http://localhost:8000/graphql.

Environment Variables¶

Schema Reference¶

Types¶

TaskType¶

The primary task representation:

type TaskType {
  id: String!
  name: String!
  status: TaskStatusEnum!
  priority: Int!
  progress: Float!
  result: String
  error: String
  createdAt: DateTime
  updatedAt: DateTime
  parentId: String
  description: String
  children: [TaskType!]
}

TaskStatusEnum¶

enum TaskStatusEnum {
  PENDING
  IN_PROGRESS
  COMPLETED
  FAILED
  CANCELLED
}

Input Types¶

CreateTaskInput¶

input CreateTaskInput {
  name: String!
  description: String
  executor: String
  priority: Int
  inputs: JSON
  parentId: String
}

UpdateTaskInput¶

input UpdateTaskInput {
  name: String
  description: String
  status: TaskStatusEnum
  inputs: JSON
}

Queries¶

task¶

Fetch a single task by ID.

query {
  task(taskId: "task-123") {
    id
    name
    status
    progress
    result
    error
    createdAt
  }
}

tasks¶

Fetch a list of tasks with optional filters.

query {
  tasks(status: PENDING, limit: 10, offset: 0) {
    id
    name
    status
    priority
    createdAt
  }
}

Arguments:

taskChildren¶

Fetch all child tasks of a parent task.

query {
  taskChildren(parentId: "parent-task-123") {
    id
    name
    status
    priority
  }
}

Mutations¶

createTask¶

Create a new task.

mutation {
  createTask(taskInput: {
    name: "my_task"
    description: "A sample task"
    executor: "rest_executor"
    priority: 5
    inputs: { url: "https://api.example.com/data" }
  }) {
    id
    name
    status
  }
}

updateTask¶

Update an existing task.

mutation {
  updateTask(
    taskId: "task-123"
    taskInput: {
      name: "updated_name"
      status: CANCELLED
    }
  ) {
    id
    name
    status
    updatedAt
  }
}

cancelTask¶

Cancel a running task. Returns true on success.

mutation {
  cancelTask(taskId: "task-123")
}

deleteTask¶

Delete a task. Returns true on success.

mutation {
  deleteTask(taskId: "task-123")
}

Subscriptions¶

taskStatusChanged¶

Subscribe to real-time status updates for a specific task. Emits a TaskType each time the task's status changes. The subscription ends automatically when the task reaches a terminal status (COMPLETED, FAILED, or CANCELLED).

subscription {
  taskStatusChanged(taskId: "task-123") {
    id
    name
    status
    progress
    result
    error
  }
}

The subscription uses the graphql-transport-ws protocol over WebSocket.

Client Examples¶

Python (httpx)¶

import httpx

# Query a task
query = """
query GetTask($taskId: String!) {
  task(taskId: $taskId) {
    id
    name
    status
    progress
    result
  }
}
"""

response = httpx.post(
    "http://localhost:8000/graphql",
    json={"query": query, "variables": {"taskId": "task-123"}},
)
print(response.json())

Python (create task)¶

import httpx

mutation = """
mutation CreateTask($input: CreateTaskInput!) {
  createTask(taskInput: $input) {
    id
    name
    status
  }
}
"""

variables = {
    "input": {
        "name": "my_executor",
        "description": "Run a data pipeline",
        "priority": 5,
        "inputs": {"source": "s3://bucket/data.csv"},
    }
}

response = httpx.post(
    "http://localhost:8000/graphql",
    json={"query": mutation, "variables": variables},
)
print(response.json())

cURL¶

# Query tasks
curl -X POST http://localhost:8000/graphql \
  -H "Content-Type: application/json" \
  -d '{
    "query": "{ tasks(status: PENDING, limit: 5) { id name status priority } }"
  }'

# Create a task
curl -X POST http://localhost:8000/graphql \
  -H "Content-Type: application/json" \
  -d '{
    "query": "mutation { createTask(taskInput: { name: \"my_task\" }) { id name status } }"
  }'

Authentication¶

The GraphQL adapter supports the same JWT authentication as the A2A and MCP adapters. Include the token in the Authorization header:

curl -X POST http://localhost:8000/graphql \
  -H "Authorization: Bearer <your-jwt-token>" \
  -H "Content-Type: application/json" \
  -d '{"query": "{ tasks { id name status } }"}'

See the API Server Guide for token generation details.

GraphiQL Playground¶

When APFLOW_ENABLE_DOCS=true (the default), opening http://localhost:8000/graphql in a browser loads the GraphiQL interactive playground. Use it to:

• Explore the schema with auto-complete
• Run queries and mutations interactively
• View type documentation inline
• Test subscriptions

To disable GraphiQL in production:

export APFLOW_ENABLE_DOCS=false

Protocol Comparison¶

See Also¶

• API Server Guide - Server setup and protocol selection
• HTTP API Reference - A2A Protocol HTTP API
• Python API Reference - Python library API
• Environment Variables - All configuration options