Build modular, testable LLM agents by composing behavior from dataclass components, async systems, and pluggable providers. No inheritance hierarchies, just clean composition.
# Clone and install with uv
git clone https://github.com/MoveCloudROY/ecs-agent.git
cd ecs-agent
uv sync --group dev
# Install with embeddings support (optional)
uv pip install -e ".[embeddings]"
# Install with MCP support (optional)
uv pip install -e ".[mcp]"Requires Python ≥ 3.11
import asyncio
import os
from ecs_agent.components import ConversationComponent, LLMComponent
from ecs_agent.core import Runner, World
from ecs_agent.providers import OpenAIProvider
from ecs_agent.systems.reasoning import ReasoningSystem
from ecs_agent.systems.memory import MemorySystem
from ecs_agent.systems.error_handling import ErrorHandlingSystem
from ecs_agent.types import Message
async def main() -> None:
world = World()
# Create a provider (any OpenAI-compatible API works)
provider = OpenAIProvider(
api_key=os.environ["LLM_API_KEY"],
base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"),
model=os.getenv("LLM_MODEL", "gpt-4o"),
)
# Create an agent entity and attach components
agent = world.create_entity()
world.add_component(agent, LLMComponent(
provider=provider,
model=provider.model,
system_prompt="You are a helpful assistant.",
))
world.add_component(agent, ConversationComponent(
messages=[Message(role="user", content="Hi there!")],
))
# Register systems (priority controls execution order)
world.register_system(ReasoningSystem(priority=0), priority=0)
world.register_system(MemorySystem(), priority=10)
world.register_system(ErrorHandlingSystem(priority=99), priority=99)
# Run the agent loop
runner = Runner()
await runner.run(world, max_ticks=3)
# Read results
conv = world.get_component(agent, ConversationComponent)
if conv:
for msg in conv.messages:
print(f"{msg.role}: {msg.content}")
asyncio.run(main())Mix 35+ components to build custom agents without inheritance bloat. The Entity-Component-System (ECS) pattern keeps logic and data strictly separated, making agents modular, serializable, and easy to test. Fully type-safe with strict mypy and dataclass(slots=True).
- Subagent Delegation — Spawn child agents for subtasks with skill and permission inheritance.
- MessageBus — Parent-child and sibling messaging via pub/sub or request-response patterns.
- Unified API — Control lifecycle with
subagent,subagent_status,subagent_result, andsubagent_canceltools.
- Tree Conversations — Branch reasoning paths, navigate multiple strategies, and linearize history for LLM compatibility.
- Planning & ReAct — Multi-step reasoning with dynamic replanning on errors or unexpected tool results.
- MCTS Optimization (experimental) — Find optimal execution paths using Monte Carlo Tree Search for complex goals.
- TaskComponent — Structured multi-step task definitions with description, expected output, agent assignment, tool lists, and context dependencies.
- Priority & Retries — Priority-based ordering and configurable retry limits for robust execution.
- Output Schema — Optional JSON schema validation for task outputs.
- Trigger Templates — Define
@keywordorevent:<name>triggers that inject pre-defined prompt blocks into user messages. - One-Shot Context Pool — Automatically collect tool results and subagent outputs into a transient context block for the next LLM turn.
- Stable Injection Order —
[PROMPT_INJECT:...]marker → trigger block → context pool block → original user text. - Transient Lifecycle — Injections are provider-call only and not persisted in conversation history. Context pool clears only on successful LLM turn.
- Markdown Skills — Define agent capabilities via
SKILL.mdfiles with YAML frontmatter. System prompts are injected automatically, and@-prefixed relative paths are resolved to workspace-safe paths at load time. - Script Skills — Extend markdown skills with Python tool handlers in a
scripts/directory, executed as sandboxed subprocesses. - Built-in Tools —
BuiltinToolsSkillprovidesread_file,write_file,edit_file, andbashwith workspace binding, path traversal protection, and hash-anchored editing. - Skill Discovery — File-based skill loading from directories with metadata-first activation and lifecycle management via
SkillManager.
- 5 LLM Providers + Streaming — OpenAI, Claude, LiteLLM (100+ models), Fake, and Retry providers with real-time SSE token delivery.
- Context Management — Checkpoints (undo/resume), conversation compaction (compression), and memory windowing.
- Tool Ecosystem — Auto-discovery via
@tooldecorator, manual approval flows, securebwrapsandboxing, and composable skills. - MCP Integration — Connect to external MCP tool servers via stdio, SSE, or HTTP transports with namespaced tool mapping.
src/ecs_agent/
├── core/
│ ├── world.py # World, entity/component/system registry
│ ├── runner.py # Runner, tick loop until TerminalComponent
│ ├── system.py # System Protocol + SystemExecutor
│ ├── component.py # ComponentStore
│ ├── entity.py # EntityIdGenerator
│ ├── query.py # Query engine for entity filtering
│ └── event_bus.py # Pub/sub EventBus
├── components/
│ └── definitions.py # 30 component dataclasses
├── providers/
│ ├── protocol.py # LLMProvider Protocol
│ ├── openai_provider.py # OpenAI-compatible HTTP provider (httpx)
│ ├── claude_provider.py # Anthropic Claude provider
│ ├── litellm_provider.py # LiteLLM unified provider
│ ├── fake_provider.py # Deterministic test provider
│ └── retry_provider.py # Retry wrapper (tenacity)
├── systems/ # 15 built-in systems
│ ├── reasoning.py # LLM inference
│ ├── planning.py # Multi-step plan execution
│ ├── replanning.py # Dynamic plan adjustment
│ ├── tool_execution.py # Tool call dispatch
│ ├── permission.py # Tool whitelisting/blacklisting
│ ├── memory.py # Conversation memory management
│ ├── collaboration.py # (Removed in favor of MessageBusSystem)
│ ├── message_bus.py # Pub/sub and request-response messaging
│ ├── error_handling.py # Error capture and recovery
│ ├── tree_search.py # MCTS plan optimization
│ ├── tool_approval.py # Human-in-the-loop approval
│ ├── rag.py # Retrieval-Augmented Generation
│ ├── checkpoint.py # World state snapshots
│ ├── compaction.py # Conversation compaction
│ ├── user_input.py # Async user input
│ └── subagent.py # Subagent delegation
├── tools/
│ ├── __init__.py # Tool utilities
│ ├── discovery.py # Auto-discovery of tools
│ ├── sandbox.py # Secure execution environment
│ ├── bwrap_sandbox.py # bwrap-backed isolation
│ ├── builtins/ # Standard library skills
│ │ ├── file_tools.py # read/write/edit logic
│ │ ├── bash_tool.py # Shell execution
│ │ ├── edit_tool.py # Hash-anchored editing core
│ │ └── __init__.py # BuiltinTools ScriptSkill definition
├── types.py # Core types (EntityId, Message, ToolCall, etc.)
├── serialization.py # WorldSerializer for save/load
└── logging.py # structlog configuration
├── skills/ # Skills system
│ ├── protocol.py # ScriptSkill Protocol definition
│ ├── manager.py # SkillManager lifecycle handler
│ ├── discovery.py # File-based skill discovery
│ └── web_search.py # Brave Search integration
├── mcp/ # MCP integration
The Runner repeatedly ticks the World until a TerminalComponent is attached to an entity. Execution also stops if max_ticks is reached (default 100). Pass max_ticks=None for infinite execution until a TerminalComponent is found. Each tick follows this flow:
- Systems execute in priority order (lower = earlier).
- Those at the same priority level run concurrently.
- Logical operations read or write components on entities. This represents the entire data flow.
World
├── Entity 0 ── [LLMComponent, ConversationComponent, PlanComponent, ...]
├── Entity 1 ── [LLMComponent, ConversationComponent, MessageBusSubscriptionComponent, ...]
└── Systems ─── [ReasoningSystem(0), PlanningSystem(0), MessageBusSystem(5), MemorySystem(10), ...]
└── Systems ─── [ReasoningSystem(0), PlanningSystem(0), ToolExecutionSystem(5), MemorySystem(10), ...]
│
Runner.run()
│
Tick 1 → Tick 2 → ... → TerminalComponent found → Done
| Component | Purpose |
|---|---|
LLMComponent |
Provider, model, system prompt |
ConversationComponent |
Message history with optional size limit |
PlanComponent |
Multi-step plan with progress tracking |
ToolRegistryComponent |
Tool schemas and async handler functions |
PendingToolCallsComponent |
Tool calls awaiting execution |
ToolResultsComponent |
Results from completed tool calls |
MessageBusConfigComponent |
Configuration for messaging (timeouts, queue sizes) |
MessageBusSubscriptionComponent |
Registry of topic subscriptions for an entity |
MessageBusConversationComponent |
Tracks active request-response conversations |
SystemPromptComponent |
Single source for system prompt assembly (template + sections) |
KVStoreComponent |
Generic key-value scratch space |
ErrorComponent |
Error details for failed operations |
TerminalComponent |
Signals agent completion |
ToolApprovalComponent |
Policy-based tool call filtering |
SandboxConfigComponent |
Execution limits for tools |
PlanSearchComponent |
MCTS search configuration |
RAGTriggerComponent |
Vector search retrieval state |
EmbeddingComponent |
Embedding provider reference |
VectorStoreComponent |
Vector store reference |
StreamingComponent |
Enables system-level streaming output |
CheckpointComponent |
Stores world state snapshots for undo |
CompactionConfigComponent |
Token threshold and model for compaction |
ConversationArchiveComponent |
Archived conversation summaries |
RunnerStateComponent |
Tracks runner tick state and pause |
UserInputComponent |
Async user input with optional timeout |
SkillComponent |
Registry of installed skills and metadata |
PermissionComponent |
Tool whitelist/blacklist for permission control |
SkillMetadata |
Tier 1 metadata for an installed skill |
MCPConfigComponent |
Configuration for MCP transport (stdio/SSE/HTTP) |
MCPClientComponent |
Active MCP client session and tool cache |
ConversationTreeComponent |
Tree-structured conversation with branching and linearization |
ResponsesAPIStateComponent |
Tracks OpenAI Responses API state and metadata |
SubagentRegistryComponent |
Registry of named subagent configurations |
TaskComponent |
Multi-step task definition and tracking |
ScratchbookRefComponent |
Reference to a scratchbook artifact |
ScratchbookIndexComponent |
Index of scratchbook artifacts |
The examples/ directory contains 24 runnable demos:
| Example | Description |
|---|---|
chat_agent.py |
Minimal agent with dual-mode provider (FakeProvider / OpenAIProvider) |
tool_agent.py |
Tool use with automatic call/result cycling |
react_agent.py |
ReAct pattern. Thought → Action → Observation loop |
plan_and_execute_agent.py |
Dynamic replanning with RetryProvider and configurable timeouts |
streaming_agent.py |
Real-time token streaming via SSE |
retry_agent.py |
RetryProvider with custom retry configuration |
multi_agent.py |
Two agents collaborating via MessageBusSystem pub/sub (dual-mode) |
structured_output_agent.py |
Pydantic schema → JSON mode for type-safe responses |
serialization_demo.py |
Save and restore World state to/from JSON |
tool_approval_agent.py |
Manual approval flow for sensitive tools |
tree_search_agent.py |
MCTS-based planning for complex goals (dual-mode) |
rag_agent.py |
Retrieval-Augmented Generation demo (dual-mode with real embeddings) |
subagent_delegation.py |
Parent agent delegates subtasks via legacy delegate and new unified subagent tools (dual-mode) |
claude_agent.py |
Native Anthropic Claude provider usage |
litellm_agent.py |
LiteLLM unified provider for 100+ models |
streaming_system_agent.py |
System-level streaming with events |
context_management_agent.py |
Checkpoint, undo, and compaction demo (dual-mode) |
skill_agent.py |
Skill system and BuiltinTools ScriptSkill (read/write/edit) lifecycle |
skill_discovery_agent.py |
File-based skill loading from folder (dual-mode) |
permission_agent.py |
Permission-restricted agent with tool filtering (dual-mode) |
skill_agent.py |
Load a SKILL.md Skill and install it on an agent (dual-mode) |
mcp_agent.py |
MCP server integration and namespaced tool usage |
agent_dsl_json.py |
Load multi-agent configuration from JSON file using Agent DSL (dual-mode) |
agent_dsl_markdown.py |
Load agent from Markdown file with YAML frontmatter using Agent DSL (dual-mode) |
Run any example:
# FakeProvider mode (no API key needed — works out of the box)
uv run python examples/chat_agent.py
uv run python examples/tool_agent.py
# Real LLM mode (set API credentials)
LLM_API_KEY=your-api-key uv run python examples/chat_agent.py
uv run python examples/react_agent.py
# RAG with real embeddings
LLM_API_KEY=your-api-key EMBEDDING_MODEL=text-embedding-3-small uv run python examples/rag_agent.pyCopy .env.example to .env and add your API credentials:
cp .env.example .envLLM_API_KEY=your-api-key-here
LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
LLM_MODEL=qwen3.5-plus
# For DashScope (Aliyun), also set:
DASHSCOPE_API_KEY=your-api-key-here
LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
LLM_MODEL=qwen3.5-plusThen use OpenAIProvider (works with any OpenAI-compatible API):
from ecs_agent.providers import OpenAIProvider
provider = OpenAIProvider(
api_key="your-api-key",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
model="qwen3.5-plus",
)Wrap with RetryProvider for automatic retries on transient failures:
from ecs_agent import RetryProvider, RetryConfig
provider = RetryProvider(
provider=OpenAIProvider(api_key="...", base_url="...", model="..."),
config=RetryConfig(max_retries=3, initial_wait=1.0, max_wait=30.0),
)# Run all tests
uv run pytest
# Run a single test file
uv run pytest tests/test_world.py
# Run tests matching a keyword
uv run pytest -k "streaming"
# Verbose output
uv run pytest -vTo verify the integration with a real LLM (e.g., DashScope), run the following command. It uses environment variables to avoid exposing API keys and skips gracefully if LLM_API_KEY is not set:
LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1 \
LLM_MODEL=qwen3.5-flash \
LLM_API_KEY="$LLM_API_KEY" \
uv run pytest tests/test_real_llm_integration.py -k "prompt" -v# Full strict type check
uv run mypy src/ecs_agent/
# Single file
uv run mypy src/ecs_agent/core/world.py- Build: hatchling
- Package manager: uv (lockfile:
uv.lock) - pytest:
asyncio_mode = "auto", async tests run without explicit event loop setup - mypy:
strict = true,python_version = "3.11"
See docs/ for detailed guides:
- Getting Started, Installation, first agent, key concepts
- Architecture, ECS pattern, data flow, system lifecycle
- Core Concepts, World, Entity, Component, System, Runner
- API Reference, Complete API surface
- Examples, Walkthrough of all 21 examples
- Components, All 27 components with usage examples
- Systems, All 14 systems with configuration details
- Providers, LLM provider protocol, built-in providers
- Streaming, SSE streaming setup and usage
- Structured Output, Pydantic schema → JSON mode
- Serialization, World state persistence
- Logging, structlog integration
- Retry, RetryProvider configuration
- Context Management, Checkpoint, undo, and compaction
- Runtime Control, Entity registry, system lifecycle, model switching, interruption, revert
- Agent DSL, Declarative agent definition and loading
- Agent Scratchbook, Persistent filesystem-backed artifact storage and indexing
- Task Orchestration, Multi-step task management and dependency resolution
- Skills, Composable capabilities and progressive disclosure
- Built-in Tools, File manipulation and shell execution
- Tool Discovery & Approval, Auto-discovery, sandbox, approval flow
- MCP Integration, Connecting to external MCP tool servers
- Web Search, Brave Search API integration
- Permissions, Tool filtering and bwrap sandboxing
- User Input, Async human-in-the-loop input
MIT