artificer-dispatcher 0.1.1

Polls project management APIs for ready tickets and spawns AI agents to work on them

artificer-dispatcher

Polls task queues, dispatches agent subprocesses, and exposes an HTTP API so agents can interact with tasks without knowing which backend is in use.

How it works

The router polls configured queues for ready tasks. When it finds one, it moves the task to an in-progress queue, spawns a subprocess (any command), and passes the task ID via template variables. The subprocess uses a local HTTP API to read task details, post comments, update fields, and move the task when done. The agent never talks to the backend directly.

Key concepts

• Backend adapters — Protocol-based (TaskAdapter, 6 methods). Ships with Planka and JSON file adapters. Implement the protocol for anything else (Jira, Trello, Linear, SQLite, SQS, etc.).
• Agent adapters — Protocol-based (AgentAdapter). Ships with a Claude adapter (session tracking, resume hints) and a default pass-through for any command.
• Routes — Map queues to commands with template variables ({task_id}, {task_name}, {task_url}).
• HTTP API — Agents hit localhost. No credentials, no backend coupling.

Quick start

Requires Python 3.13+.

pip install -e .

Add configuration to pyproject.toml:

[tool.artificer-dispatcher]
poll_interval = 30
max_concurrent_agents = 3

[tool.artificer-dispatcher.backend]
type = "json"
url = "/tmp/board.json"

[tool.artificer-dispatcher.routing."My Queue"]
command = "my-agent"
args = ["--task", "{task_id}", "--name", "{task_name}"]
in_progress_queue = "In Progress"

artificer-dispatcher           # installed console script
python -m artificer_dispatcher # module invocation
python main.py                 # development shortcut

CLI flags

Flag	Description
--debug	Enable debug logging
--port PORT	Override api_port from config

This starts two things:

1. Router — polls configured queues, picks up tasks, moves them to in-progress, and spawns agent subprocesses.
2. HTTP API — listens on http://{api_host}:{api_port} so spawned agents can interact with tasks.

Configuration reference

All config lives in pyproject.toml under [tool.artificer-dispatcher].

Top-level settings

[tool.artificer-dispatcher]
poll_interval = 30            # seconds between polls (default: 30)
max_concurrent_agents = 3     # max agent processes at once (default: 3)
default_agent_timeout = 3600  # default timeout in seconds for all agents (optional)
api_host = "127.0.0.1"       # HTTP API bind address (default: 127.0.0.1)
api_port = 8000              # HTTP API port (default: 8000)

Backend config

[tool.artificer-dispatcher.backend]
type = "planka"                  # "planka" or "json"
url = "http://localhost:1337"    # Planka server URL or path to .json file

Route config

Each routing section maps a queue to a command:

[tool.artificer-dispatcher.routing."Queue Name"]
command = "my-agent"
args = ["--task", "{task_id}", "{task_name}"]
in_progress_queue = "In Progress"
timeout = 1800  # route-specific timeout in seconds (optional, overrides default)

Template variables

These placeholders are substituted in args when spawning an agent:

Variable	Description
{task_id}	Task/card ID
{task_name}	Task/card title
{task_url}	Link to the task (e.g. Planka card URL)

Agent timeouts

You can configure timeouts to automatically terminate agent processes that run too long:

• default_agent_timeout (optional): Sets a global timeout in seconds for all agents. If not specified, agents run indefinitely.
• timeout (optional, per-route): Sets a route-specific timeout in seconds. Overrides default_agent_timeout for that route.

When an agent times out:

1. The process receives a TERM signal and has 5 seconds to exit gracefully
2. If it doesn't exit, it receives a KILL signal
3. A comment is added to the task noting the timeout

Example:

[tool.artificer-dispatcher]
default_agent_timeout = 3600  # 1 hour default for all agents

[tool.artificer-dispatcher.routing."Quick Tasks"]
command = "my-agent"
args = ["--task", "{task_name}"]
timeout = 300  # 5 minutes for quick tasks (overrides default)

[tool.artificer-dispatcher.routing."Long Tasks"]
command = "my-agent"
args = ["--task", "{task_name}"]
# No timeout specified - uses default of 3600 seconds

Authentication

For the Planka backend, set one of these in your environment (or .env):

# Option 1: API token
PLANKA_TOKEN=your-token-here

# Option 2: Username + password
PLANKA_USER=admin
PLANKA_PASSWORD=secret

HTTP API

Method	Endpoint	Description
GET	/tasks/{task_id}	Full task info: description, labels, assignees, comments
POST	/tasks/{task_id}/comments	Post a comment on a task ({"comment": "text"})
POST	/tasks/{task_id}/move	Move a task to a different queue ({"target_queue": "name"})
PATCH	/tasks/{task_id}	Update task fields ({"name": "...", "description": "...", "labels": [...], "assignees": [...]})
POST	/tasks	Create a new task ({"queue_name": "...", "name": "...", "description": "..."})
GET	/status	Router status: active agents, available slots

Task lifecycle

1. Task sits in a watched queue (e.g. Todo)
2. Router picks it up, moves it to the in-progress queue, and assigns the authenticated user
3. Router spawns the configured command as a subprocess
4. The agent uses the HTTP API to read task details, add comments, etc.
5. When finished, the agent calls the move endpoint to move the task to a done queue

Backends

Planka

Uses dot-notation for queue naming: Project.Board.List.

[tool.artificer-dispatcher.backend]
type = "planka"
url = "http://localhost:1337"

[tool.artificer-dispatcher.routing."My Project.My Board.Todo"]
command = "my-agent"
args = ["-p", "Work on task {task_id}: {task_name}"]
in_progress_queue = "My Project.My Board.In Progress"

JSON file

For development/testing or lightweight use without external services.

[tool.artificer-dispatcher.backend]
type = "json"
url = "/tmp/board.json"

The JSON file structure:

{
  "queues": {
    "Todo": [
      {"id": "1", "name": "Fix crash", "description": "...", "labels": [], "assignees": [], "comments": [], "tasks": []}
    ],
    "In Progress": [],
    "Done": []
  }
}

Custom

Implement the TaskAdapter protocol (6 methods) in artificer_dispatcher/adapters/base.py:

• get_ready_tasks(queue_names) — Return tasks from the given queues
• get_task(task_id) — Return a single task by ID
• move_task(task_id, target_queue) — Move a task between queues
• add_comment(task_id, text) — Add a comment to a task
• update_task(task_id, *, assignees, name, description, labels) — Update task fields
• create_task(queue_name, name, description) — Create a new task

Development

pip install -e ".[dev]"
pytest