Symphony Python Port: Development Plan for Porting an Elixir AI Agent Orchestrator to Python 3.12

Forum topic · ✨步子哥 · 2026-03-15

Summary

This post presents a detailed development plan for porting Symphony, an Elixir-based AI agent orchestration system, 1:1 to Python 3.12 using AgentScope as the agent foundation library. The document maps every Elixir module (Orchestrator, AgentRunner, Workflow, Linear client, Workspace, PromptBuilder, Codex AppServer, dashboard, etc.) to a corresponding Python implementation, replacing Elixir/OTP concurrency (GenServer, Process.monitor) with asyncio primitives, dataclasses, and async locks. Configuration is validated via Pydantic v2 models covering tracker, polling, workspace, hooks, agent, Codex, and server settings. The post specifies a full project structure, a technology stack (httpx, FastAPI, Jinja2, APScheduler, Rich, structlog, asyncssh, watchfiles), and an AgentScope integration strategy that uses low-level components (Msg, ModelWrapper, Toolkit) instead of the built-in ReActAgent for fine-grained lifecycle control. Work is organized into nine phases: skeleton and config, Linear tracker and models, workspace management, agent and prompt integration, the orchestrator core, dashboard and logging, optional FastAPI web API, CLI and integration, and testing/documentation, with acceptance criteria including >90% type annotations and >80% test coverage plus a risk assessment table.

This document is a comprehensive development plan for porting Symphony, originally written in Elixir, to Python 3.12 using AgentScope as the agent foundation library. The Python project lives in symphony.py/, a 1:1 port of the Elixir source in symphony/.

Architecture Mapping

Each Elixir module maps to a Python counterpart:

| Elixir Module | Responsibility | Python Implementation | |------|----------|-----------------| | Orchestrator | Core scheduler, polling and task dispatch | symphony.orchestrator.Orchestrator (asyncio) | | AgentRunner | Runs the Agent for a single Issue | symphony.agent_runner.AgentRunner | | Workflow | Loads WORKFLOW.md | symphony.workflow.WorkflowLoader | | Config / Config.Schema | Configuration and schema validation | symphony.config with Pydantic models | | Linear.Client / Linear.Issue | Linear GraphQL API client and Issue model | symphony.trackers.linear_client.LinearClient, symphony.models.issue.Issue | | Workspace | Workspace management | symphony.workspace.WorkspaceManager | | PromptBuilder | Prompt construction | symphony.prompts.PromptBuilder | | Codex.AppServer / Codex.DynamicTool | Codex protocol and dynamic tools | AgentScopeClient, LinearGraphQLTool | | StatusDashboard, HttpServer, PathSafety, SSH | Dashboard, web API, path safety, remote execution | Dashboard, WebServer (FastAPI), PathSafety, SSHClient |

The core data flow: a WorkflowLoader feeds the asyncio-based Orchestrator, which coordinates a Linear Client, Agent Runner, and Workspace Manager. The Agent Runner drives an AgentScope ReActAgent with custom tools, with an observability layer (logging, dashboard, FastAPI HTTP API) on top.

Project Structure

The proposed layout uses a src/symphony/ package with submodules for config/, models/, workflow/, trackers/, workspace/, agents/ (with tools/), orchestrator/, dashboard/, web/, and utils/, plus tests/ (unit, integration, fixtures) and Chinese documentation (架构说明.md, 配置指南.md, etc.). Entry point is python -m symphony.

Technology Stack

Key dependencies:

Python ^3.12, agentscope ^1.0
httpx ^0.27, fastapi ^0.115, pydantic ^2.9, pydantic-settings ^2.5
jinja2 ^3.1, apscheduler ^3.10, rich ^13.9, structlog ^24.4
pyyaml ^6.0, asyncssh ^2.17, watchfiles ^0.24, pytest-asyncio ^0.24

AgentScope Integration Strategy

Because Symphony needs fine-grained control over agent lifecycle and interaction with the Codex app-server:

1. Use AgentScope's low-level components (Msg, ModelWrapper, Toolkit) 2. Define a custom SymphonyAgent inheriting from AgentBase 3. Register linear_graphql as an AgentScope tool 4. Do not use the built-in ReActAgent directly, since Symphony must control multi-turn dialogue and state itself

Core Module Design

The configuration system uses Pydantic models, including TrackerConfig (Linear/memory tracker with endpoint, active/terminal states), PollingConfig (default 30s interval), WorkspaceConfig, HooksConfig, AgentConfig (max 10 concurrent agents, 20 turns), CodexConfig (command, approval policy, sandbox, timeouts), and ServerConfig.

The Issue model normalizes Linear issues with fields such as identifier, title, state, branch_name, labels, and blocked_by references.

The orchestrator state machine uses dataclasses: RunningEntry tracks each running issue (task, session, token counts, retries), RetryEntry backs the retry queue with due timestamps, and OrchestratorState aggregates running tasks, claimed issues, retry attempts, completed issues, and Codex token totals.

Implementation Phases

1. Project skeleton and configuration — directory structure, pyproject.toml, Pydantic config, WORKFLOW.md loader, unit tests 2. Issue Tracker and models — Issue model, tracker base class, Linear GraphQL client, in-memory tracker for tests, pagination and normalization 3. Workspace management — path safety, lifecycle management, hooks, local and optional SSH workspaces 4. Agent and prompts — prompt builder, Linear GraphQL tool, ReAct agent, AgentScope client integration 5. Orchestrator core — state, dispatch, retry with backoff, reconciliation, polling loop 6. Dashboard and logging — structured logging, snapshots, Rich terminal dashboard with real-time refresh 7. Web API (optional) — FastAPI server with GET /api/v1/state, GET /api/v1/<issue_id>, POST /api/v1/refresh 8. CLI and integration — argument parsing, startup flow, hot config reload, graceful shutdown, integration tests 9. Testing and documentation — unit tests (>80% coverage), README and full docs, code review

Key Technical Decisions

Async architecture: asyncio replaces Elixir OTP. GenServer maps to asyncio.Queue/asyncio.Event, Process.monitor to task exception handling via add_done_callback, and Process.send_after to asyncio.call_later.
State management: dataclasses protected by asyncio.Lock.
AgentScope integration: low-level wrappers and Toolkit registration instead of the high-level ReActAgent.

Acceptance Criteria

Full WORKFLOW.md parsing, Linear connectivity, workspace creation with hooks, agent dispatch, multi-turn dialogue, retry with backoff, correct state reconciliation, real-time dashboard
Configurable polling (default 30s) and concurrency (default 10 agents), stable memory with no leaks
>90% type annotation coverage, >80% unit test coverage, PEP8 compliance, complete docstrings

Risks

| Risk | Impact | Mitigation | |------|------|----------| | AgentScope API changes | High | Pin versions, monitor updates | | Linear API changes | Medium | Isolate the API layer | | Python asyncio complexity | Medium | Thorough testing, code review | | SSH remote execution security | High | Strict input validation |

Appendix

References: AgentScope documentation, Symphony SPEC.md, and the original Elixir source. A terminology glossary maps Elixir concepts (GenServer, Agent, handle_info) to their Python equivalents (asyncio + dataclass, AgentRunner + SymphonyAgent, message-handling coroutines).

*Document version 1.0, last updated 2026-03-15.*