Agentex

是什么

Agentex 是把 AI 智能体（AI Agent）变成可注册、可编排、可观测的后端服务的一套平台范式。 用它的效果是：智能体不再是一段散落的脚本，而是带状态、带消息、带工作流的产品功能。

怎么用

1. 先按业务场景挑选合适的智能体类型（对话型、工作流型、批处理型），让后续设计有清晰边界。
2. 在 acp.py 与 manifest.yaml 里把入口、状态、工具声明清楚，让平台能正确注册并调度这个智能体。
3. 用 adk.messages（消息接口）传输用户输入与系统响应，让对话过程可追溯、可重放。
4. 用 adk.state（状态接口）持久化业务上下文，让多轮交互不会因为重启而丢失记忆。
5. 在本地拉起 FastAPI 后端与 Temporal 工作流后，再把智能体跑起来做端到端验证。

架构图

flowchart LR
  用户输入 --> 智能体注册
  智能体注册 --> 工作流编排
  工作流编排 --> 状态存储
  工作流编排 --> 消息接口
  消息接口 --> 用户输入

Agentex Platform

Agentex is a platform for building and deploying intelligent agents. The repo has two main parts:

• agentex/ — FastAPI backend + Temporal workflows (runs in Docker)
• agentex-ui/ — Next.js frontend (runs locally)

Agents are built with the agentex-sdk CLI and run as separate processes that register with the backend.

When to Activate

• Choosing between sync, async, or Temporal agent type for a new agent
• Wiring acp.py, manifest.yaml, or run_worker.py for a new agent
• Using adk.messages, adk.state, or adk.providers in an activity or workflow
• Debugging ACP protocol issues or agent registration failures
• Understanding the backend DDD layer boundaries or exception mapping
• Windows-specific setup issues (uv sync, port conflicts, .env loading)

---

Agent Types

Sync ACP

One message in, one response out. Stateless.

acp = FastACP.create(acp_type="sync")

@acp.on_message_send
async def handle(params: SendMessageParams) -> TaskMessageContent:
    return TextContent(author="agent", content="reply")

Use when: FAQ bots, translation, data lookups, single-turn interactions.

Async ACP (base)

Task lifecycle with persistent state across multiple turns.

acp = FastACP.create(acp_type="async", config=AsyncACPConfig(type="base"))

@acp.on_task_create   # called once — initialize state
@acp.on_task_event_send  # called per message — respond via adk.messages.create
@acp.on_task_cancel   # called on cancel — cleanup

Key difference from sync: responses are pushed via adk.messages.create, not returned. State is persisted via adk.state.create / get_by_task_and_agent / update.

Use when: multi-turn conversations, stateful workflows, streaming LLM responses. Warning: race conditions if parallel events arrive — use Temporal for production.

Async ACP + Temporal

Same as Async but every step is a durable Temporal workflow. Survives crashes and restarts.

# manifest.yaml
agent:
  acp_type: async
  temporal:
    enabled: true

Use when: production agents, long-running tasks, human-in-the-loop, complex multi-step tool chains.

---

ACP State Pattern (Async)

class MyState(BaseModel):
    turn: int
    messages: List[Message]

# Create on task init
await adk.state.create(task_id=..., agent_id=..., state=MyState(...))

# Read on each event
task_state = await adk.state.get_by_task_and_agent(task_id=..., agent_id=...)
state = MyState.model_validate(task_state.state)

# Write back after mutating
await adk.state.update(state_id=task_state.id, task_id=..., agent_id=..., state=state)

---

Sending Messages (Async)

# Echo user message back (so it shows in UI)
await adk.messages.create(task_id=params.task.id, content=params.event.content)

# Send agent reply
await adk.messages.create(
    task_id=params.task.id,
    content=TextContent(author="agent", content="response text"),
)

# Streaming LLM (auto-sends chunks to UI)
await adk.providers.litellm.chat_completion_stream_auto_send(
    task_id=params.task.id,
    llm_config=LLMConfig(model="gpt-4o-mini", messages=state.messages, stream=True),
)

---

manifest.yaml Structure

local_development:
  agent:
    port: 8000          # must be unique per agent (8000, 8001, 8002...)
    host_address: host.docker.internal
  paths:
    acp: project/acp.py

agent:
  name: my-agent        # unique name, shown in UI
  acp_type: sync        # or async
  temporal:
    enabled: false
  credentials: []
  env: {}

---

Backend Architecture

src/
├── api/routes/         # FastAPI endpoints
├── domain/entities/    # Pure Pydantic models
├── domain/use_cases/   # Business logic
├── adapters/crud_store/ # DB adapters (Postgres + MongoDB)
├── adapters/streams/   # Redis SSE streams
└── config/dependencies.py  # Singleton GlobalDependencies

Layer rules:

• Domain layer has zero framework imports
• API layer → use cases → domain ← adapters
• ORM ↔ domain conversion via explicit converter functions — never skip layers

Exceptions:

• ClientError → 400, ServiceError → 500, ItemDoesNotExist → 404

---

Windows-Specific Gotchas

Problem	Fix
uv sync fails: platform not compatible	Add "sys_platform == 'win32'" to environments in root pyproject.toml, then uv lock
load_dotenv(override=True) clobbers Docker env vars	Change to override=False in environment_variables.py
Local PostgreSQL on port 5432 blocks Docker	Change Docker postgres port to 5434:5432 in docker-compose.yml
agentex init Unicode error	Set $env:PYTHONUTF8 = "1" before running
agentex init path has \n in it	Type short relative name (my-agent), not a full path
source .venv/bin/activate fails	Use .venv\Scripts\Activate.ps1 on Windows
Temporal worker connects to localhost inside Docker	Caused by .env overriding Docker network hostnames — needs override=False

---

Ports

Port	Service
3000	Frontend UI
5003	FastAPI backend (Swagger at /swagger)
5432	Local PostgreSQL (if installed)
5434	Docker agentex-postgres (remapped to avoid conflict)
5433	Docker Temporal PostgreSQL
6379	Redis
7233	Temporal server
8080	Temporal UI
8000+	Agent ACP servers (one port per agent)
27017	MongoDB

---

Key Environment Variables (agentex/.env)

ENVIRONMENT=development
DATABASE_URL=postgresql://postgres:[email protected]:5434/agentex
TEMPORAL_ADDRESS=localhost:7233
REDIS_URL=redis://localhost:6379
MONGODB_URI=mongodb://localhost:27017
MONGODB_DATABASE_NAME=agentex
AGENTEX_SERVER_TASK_QUEUE=agentex-server
ALLOWED_ORIGINS=http://localhost:3000
ENABLE_HEALTH_CHECK_WORKFLOW=true

---

Running Tests

cd agentex
# Unit tests (no Docker needed)
.\build.ps1 test-unit

# Integration tests (needs Docker infra running)
.\build.ps1 test-integration

# Specific file
.\build.ps1 test -File tests/unit/test_foo.py

---

Red Flags

• Sync ACP for multi-turn conversations — sync agents receive one message and return one reply; they have no state, no turn history, and no mechanism to stream responses; use async ACP (or async + Temporal) for any stateful interaction
• Handler decorators in acp.py for a Temporal agent — Temporal agents route all ACP events through the workflow engine; registering @acp.on_task_create decorators in acp.py bypasses Temporal and runs handlers outside the durable execution context
• Returning a response from an async handler instead of using adk.messages.create — async agent handlers are not expected to return a value; the return value is silently discarded and the user sees no reply; push responses explicitly via adk.messages.create
• Not following load → mutate → save with adk.state — reading state, mutating it in-memory, and then returning without saving means the next signal handler loads stale state; always call adk.state.update after every mutation before returning
• Omitting get_all_activities() in run_worker.py — ADK built-in activities (messages, state persistence, tracing) are registered via get_all_activities(); omitting it means all adk.messages.create and adk.state.* calls fail at runtime with "activity not found"
• Two agents sharing the same port in manifest.yaml — each ACP server process binds a port; running two agents with the same local_development.agent.port causes one to fail to start; increment the port for each agent (8000, 8001, 8002, …)
• load_dotenv(override=True) when running inside Docker — overriding with the local .env file replaces Docker-injected environment variables such as DATABASE_URL and TEMPORAL_ADDRESS with localhost values, breaking service discovery inside the container network

Checklist

• acp_type chosen correctly in manifest.yaml (sync / async / async + temporal)
• Temporal agent acp.py has only FastACP.create(acp_type="async", config=TemporalACPConfig(...)) — no handler decorators
• adk.messages.create used to send responses (not returned from handlers)
• State follows load → mutate → save pattern via adk.state
• on_task_create ends with await workflow.wait_condition(lambda: self._done) for Temporal agents
• get_all_activities() included in worker alongside custom activities
• Agent port in manifest.yaml is unique across all running agents (8000, 8001, …)
• Windows: load_dotenv(override=False) to avoid clobbering Docker env vars
• Domain exceptions (ClientError, ServiceError, ItemDoesNotExist) used — not HTTPException in use cases