Project Structure Guide

Learn about the files and folders in your Agentex agent project and what each one does.

What You Get With agentex init

When you run agentex init, the CLI creates a complete agent project with everything you need to get started. The exact structure depends on which agent type you choose:

• Sync Agent - Simple request-response patterns
• Async Base Agent - Custom async implementations
• Async Temporal Agent - Complex, durable workflows

This guide is organized by agent type. Jump to your section:

• Sync Agent Structure
• Async Base Agent Structure
• Async Temporal Agent Structure
• Common Files

---

Sync Agent Structure

Project Layout

my-sync-agent/
├── project/
│   └── acp.py               # ACP server with message handler
├── manifest.yaml            # Agent configuration
├── Dockerfile               # Container definition
├── .dockerignore           # Docker build exclusions
├── dev.ipynb               # Development notebook
└── pyproject.toml          # Dependencies (UV)

Sync-Specific Files

project/acp.py

Your ACP server defines how the agent responds to incoming messages. For sync agents, you handle messages with a simple decorator and return a response immediately - no workflows, no activities, just direct request-response.

from agentex.lib.sdk.fastacp.fastacp import FastACP
from agentex.lib.types.acp import SendMessageParams
from agentex.types.task_message_content import TaskMessageContent
from agentex.types.text_content import TextContent

# Create a sync ACP server
acp = FastACP.create(acp_type="sync")

@acp.on_message_send
async def handle_message_send(
    params: SendMessageParams
) -> TaskMessageContent:
    """Handle incoming messages and return immediate response"""
    user_message = params.content.content

    # Your agent logic here
    # Example: Echo the message back
    response = f"You said: {user_message}"

    return TextContent(
        author="agent",
        content=response
    )

manifest.yaml

The manifest configures your agent for both local development and deployment. When you run agentex init, this file is automatically generated with the correct configuration for your agent type.

agent:
  acp_type: sync  # "sync" not "agentic"
  name: my-sync-agent
  description: "Simple sync agent"

local_development:
  agent:
    port: 8000
  paths:
    acp: project/acp.py  # No worker path needed

For customization options, see the Common Files section below or the Agent Configuration Files documentation.

---

Async Base Agent Structure

Project Layout

my-base-agent/
├── project/
│   └── acp.py               # ACP server with event handlers
├── manifest.yaml
├── Dockerfile
├── .dockerignore
├── dev.ipynb
└── pyproject.toml

Base-Specific Files

project/acp.py

The base async ACP server gives you a starting point for custom async implementations. You manually register handlers for task creation, events, and cancellation. This is useful for prototyping or when you need custom async patterns without Temporal's infrastructure.

from agentex.lib.sdk.fastacp.fastacp import FastACP
from agentex.lib.types.fastacp import AgenticBaseACPConfig
from agentex.lib.types.acp import CreateTaskParams, SendEventParams, CancelTaskParams

# Create a base agentic ACP server
acp = FastACP.create(
    acp_type="agentic",
    config=AgenticBaseACPConfig(type="base")
)

# Register your event handlers
@acp.on_task_create
async def handle_task_create(params: CreateTaskParams):
    """Called when a new task is created"""
    # Your initialization logic
    # Set up state, send welcome messages, etc.
    pass

@acp.on_task_event_send
async def handle_task_event_send(params: SendEventParams):
    """Called when an event is sent to the task"""
    # Your event handling logic
    # Process messages, call APIs, etc.
    pass

@acp.on_task_cancel
async def handle_task_cancel(params: CancelTaskParams):
    """Called when a task is canceled"""
    # Your cleanup logic
    # Close connections, save state, etc.
    pass

Unlike sync agents where everything happens in one handler, base async agents separate concerns into three handlers. You're responsible for managing state, handling async operations, and implementing retry logic yourself.

manifest.yaml

The agentex init command automatically generates this file with the correct configuration. For base async agents, there's no temporal section since you're not using Temporal workflows.

agent:
  acp_type: agentic
  name: my-base-agent
  description: "Base async agent"

local_development:
  agent:
    port: 8000
  paths:
    acp: project/acp.py  # No worker path

For customization options, see the Common Files section below or the Agent Configuration Files documentation.

---

Async Temporal Agent Structure

Project Layout

my-temporal-agent/
├── project/
│   ├── acp.py               # ACP server (minimal for Temporal)
│   ├── workflow.py          # Temporal workflow definitions
│   ├── run_worker.py        # Temporal worker setup
│   └── activities.py        # Custom activities (optional)
├── manifest.yaml            # Agent configuration
├── Dockerfile               # Container definition
├── .dockerignore           # Docker build exclusions
├── dev.ipynb               # Development notebook
└── pyproject.toml

Temporal-Specific Files

project/acp.py

For Temporal agents, the ACP server is minimal - it just forwards events to your Temporal workflows. You don't register handlers manually; the Temporal integration handles routing automatically. When a task is created or an event is sent, it's passed directly to your workflow via Temporal's infrastructure.

import os
from agentex.lib.sdk.fastacp.fastacp import FastACP
from agentex.lib.types.fastacp import TemporalACPConfig

# Create a Temporal ACP server
acp = FastACP.create(
    acp_type="agentic",
    config=TemporalACPConfig(
        type="temporal",
        temporal_address=os.getenv("TEMPORAL_ADDRESS", "localhost:7233")
    )
)

# No handlers needed - workflows handle events automatically!

Change temporal_address if you're connecting to a remote Temporal server (for example, in production or when using Temporal Cloud).

project/workflow.py

Your workflow is where all the agent logic lives. It's a class that Temporal manages, keeping it running indefinitely and ensuring it survives crashes. The workflow handles task creation and processes events (like user messages) through signals.

import json
from temporalio import workflow
from agentex.lib import adk
from agentex.lib.types.acp import CreateTaskParams, SendEventParams
from agentex.lib.core.temporal.workflows.workflow import BaseWorkflow
from agentex.lib.core.temporal.types.workflow import SignalName
from agentex.types.text_content import TextContent
from agentex.lib.environment_variables import EnvironmentVariables

environment_variables = EnvironmentVariables.refresh()

@workflow.defn(name=environment_variables.WORKFLOW_NAME)
class MyWorkflow(BaseWorkflow):
    """Durable workflow that handles tasks and events"""

    def __init__(self):
        super().__init__(display_name=environment_variables.AGENT_NAME)
        self._complete_task = False

    @workflow.signal(name=SignalName.RECEIVE_EVENT)
    async def on_task_event_send(self, params: SendEventParams) -> None:
        """Handle incoming events (user messages)"""
        # Echo user message back to the UI
        await adk.messages.create(
            task_id=params.task.id,
            content=params.event.content
        )

        # Your agent logic here
        # Call activities for external operations
        response = await self.process_message(params.event.content)

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

    @workflow.run
    async def on_task_create(self, params: CreateTaskParams) -> str:
        """Called once when task is created"""
        # Send welcome message
        await adk.messages.create(
            task_id=params.task.id,
            content=TextContent(
                author="agent",
                content="Hello! I'm ready to help."
            )
        )

        # Wait indefinitely for events (signals)
        # Workflow runs until explicitly completed
        await workflow.wait_condition(
            lambda: self._complete_task,
            timeout=None  # Can run for days/weeks/months
        )
        return "Task completed"

    async def process_message(self, content) -> str:
        """Your agent logic - call activities here for external operations"""
        # Example: call an activity
        # result = await workflow.execute_activity(
        #     "my_activity",
        #     args,
        #     start_to_close_timeout=timedelta(minutes=10)
        # )
        return "Processed message"

The @workflow.defn decorator registers your workflow with Temporal - the name must match what's in your manifest.yaml. The @workflow.run method is called once when a task is created and keeps the workflow alive. The @workflow.signal method handles incoming events like user messages.

Environment variables are automatically set from your manifest:

• WORKFLOW_NAME - from temporal.workflows[0].name
• AGENT_NAME - from agent.name
• WORKFLOW_TASK_QUEUE - from temporal.workflows[0].queue_name

project/run_worker.py

The worker connects to Temporal and executes your workflows and activities. When you run this script, it registers your workflow class and activities, then listens for work on the task queue. As tasks come in, the worker executes them using the registered workflows and activities.

import asyncio
from agentex.lib.core.temporal.activities import get_all_activities
from agentex.lib.core.temporal.workers.worker import AgentexWorker
from agentex.lib.environment_variables import EnvironmentVariables
from project.workflow import MyWorkflow

environment_variables = EnvironmentVariables.refresh()

async def main():
    task_queue_name = environment_variables.WORKFLOW_TASK_QUEUE

    # Get built-in activities + add your custom ones
    all_activities = get_all_activities() + []  # Add custom: + [my_activity]

    # Create worker
    worker = AgentexWorker(task_queue=task_queue_name)

    # Run worker with workflows and activities
    await worker.run(
        activities=all_activities,
        workflow=MyWorkflow,
    )

if __name__ == "__main__":
    asyncio.run(main())

To add custom activities, import them and add to the list:

from project.activities import my_custom_activity, another_activity

all_activities = get_all_activities() + [
    my_custom_activity,
    another_activity
]

project/activities.py

Activities are functions for non-deterministic operations - anything that could fail, take time, or return different results on retries. You use them for API calls, database queries, file operations, and heavy computations. Activities run outside the workflow, so Temporal can retry them independently if they fail.

from datetime import timedelta
from temporalio import activity
from agentex.lib.utils.logging import make_logger

logger = make_logger(__name__)

@activity.defn(name="fetch_user_data")
async def fetch_user_data(user_id: str) -> dict:
    """Fetch user data from external API"""
    logger.info(f"Fetching data for user {user_id}")

    # External API call - non-deterministic operation
    response = await httpx.get(f"https://api.example.com/users/{user_id}")
    return response.json()

Call activities from your workflow:

# In workflow.py
from datetime import timedelta
from temporalio.common import RetryPolicy

result = await workflow.execute_activity(
    "fetch_user_data",
    user_id,
    start_to_close_timeout=timedelta(minutes=10),  # Required!
    heartbeat_timeout=timedelta(minutes=1),
    retry_policy=RetryPolicy(maximum_attempts=3)
)

manifest.yaml

The Temporal manifest includes workflow configuration that specifies which workflow class to use and what task queue it listens on. The worker path tells the CLI how to start your Temporal worker for local development.

agent:
  acp_type: agentic
  name: my-temporal-agent
  description: "Temporal workflow agent"

  # Temporal configuration
  temporal:
    enabled: true
    workflows:
      - name: MyWorkflow  # Must match @workflow.defn decorator
        queue_name: my_agent_task_queue

local_development:
  agent:
    port: 8000
  paths:
    acp: project/acp.py
    worker: project/run_worker.py  # Worker path required for Temporal

The workflow name must exactly match the name in your @workflow.defn decorator. The queue_name is how Temporal routes work to your worker.

---

Common Files (All Agent Types)

These files are the same across all agent types.

manifest.yaml

The manifest.yaml is your agent's central configuration file. It controls everything from how your agent is built and packaged to how it runs locally and in production. The agentex init command generates this file automatically, but understanding each field helps you customize your agent for specific needs.

For more advanced deployment configuration, see the Agent Configuration Files documentation.

# Build Configuration
build:
  context:
    # Type: String (path)
    # Root directory for the Docker build context
    # Default: ../ (parent directory of your agent)
    # When to change: Rarely needed; only if you have a custom project structure
    root: ../

    # Type: List of strings (paths)
    # Directories/files to include in the Docker build context
    # Default: Your agent's directory name
    # When to change: Add paths if you have shared code outside your agent directory
    # Example: ["my-agent", "../shared-lib"]
    include_paths:
      - my-agent

    # Type: String (path)
    # Path to your agent's Dockerfile (relative to root)
    # Default: {agent-name}/Dockerfile
    # When to change: Only if you have a custom Dockerfile location
    dockerfile: my-agent/Dockerfile

    # Type: String (path)
    # Path to .dockerignore file (filters build context)
    # Default: {agent-name}/.dockerignore
    # When to change: Only if you have a custom .dockerignore location
    dockerignore: my-agent/.dockerignore

# Local Development Configuration
local_development:
  agent:
    # Type: Integer
    # Port where your ACP server runs locally
    # Default: 8000
    # When to change: If port 8000 is already in use
    port: 8000

    # Type: String
    # Host address for Docker networking
    # Options: "host.docker.internal" (Docker), "localhost" (direct)
    # Default: host.docker.internal
    # When to change: Use "localhost" if not running services in Docker
    host_address: host.docker.internal

  paths:
    # Type: String (path)
    # Path to your ACP server file (relative to agent root)
    # Default: project/acp.py
    # When to change: Only if you have a custom project structure
    acp: project/acp.py

    # Type: String (path)
    # Path to Temporal worker file (Temporal agents only)
    # Default: project/run_worker.py
    # When to change: Only if you have a custom project structure
    worker: project/run_worker.py

# Agent Configuration
agent:
  # Type: String
  # Defines the agent protocol type
  # Options: "sync" (request-response), "agentic" (async with events)
  # Required: Yes
  # When to change: You can change this if you update your project files to match (see agent type sections above),
  # but it's usually easier to run agentex init again and copy your code over
  acp_type: agentic

  # Type: String
  # Unique identifier for your agent
  # Required: Yes
  # Used for: Task routing, monitoring, identification
  # Example: "weather-assistant"
  name: my-agent

  # Type: String
  # Human-readable description of what your agent does
  # Required: Yes
  # Used for: Documentation, UI display
  description: "Agent description"

  # Temporal Configuration (async temporal agents only)
  temporal:
    # Type: Boolean
    # Enable Temporal workflow execution
    # Default: true for async temporal agents
    # When to use: Always true for Temporal agents, omit for base/sync
    enabled: true

    workflows:
      # Type: String
      # Name of the workflow class
      # Required: Yes (if using Temporal)
      # Must match: The name in your @workflow.defn decorator
      # Example: "MyWorkflow"
      - name: MyWorkflow

        # Type: String
        # Temporal task queue for routing work to your worker
        # Required: Yes (if using Temporal)
        # Convention: {agent-name}_task_queue
        # Example: "weather_assistant_task_queue"
        queue_name: my_agent_task_queue

    # Type: Integer
    # Port for Temporal worker health checks
    # Default: 80
    # Optional: Yes
    # When to change: If you need a different health check port
    health_check_port: 80

  # Credentials (deployment only - use .env locally)
  # Type: List of credential mappings
  # Purpose: Map Kubernetes secrets to environment variables
  # Used for: API keys, database URLs, auth tokens
  credentials:
    # Type: String
    # Name of the environment variable in your code
    # Example: "OPENAI_API_KEY"
    - env_var_name: OPENAI_API_KEY

      # Type: String
      # Name of the Kubernetes secret
      # Example: "openai-secret"
      secret_name: openai-secret

      # Type: String
      # Key within the secret to extract
      # Example: "api-key"
      secret_key: api-key

  # Type: Object (key-value pairs)
  # Set environment variables for your agent
  # Used for: Configuration that applies to both local and deployed agents
  env:
    CUSTOM_VAR: "value"
    API_TIMEOUT: "30"
    LOG_LEVEL: "info"

# Deployment Configuration
deployment:
  image:
    # Type: String
    # Container registry path for your agent's image
    # Required: Yes (for deployment)
    # Format: {registry}/{organization}/{image-name}
    # Example: "gcr.io/my-org/weather-agent"
    repository: "registry.io/my-agent"

    # Type: String
    # Image tag for versioning
    # Default: "latest"
    # Best practice: Use semantic versions in production ("1.0.0")
    tag: "latest"

  # Type: String
  # Name of Kubernetes secret for pulling private images
  # Required: If using private registry
  # Example: "my-registry-secret"
  imagePullSecrets:
    - name: my-registry-secret

---

pyproject.toml

Manage your agent's Python dependencies using UV and pyproject.toml.

[project]
name = "my-agent"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "agentex-sdk",
    "temporalio",  # For Temporal agents only
]

[project.optional-dependencies]
dev = [
    "pytest",
    "black",
    "debugpy",
]

Add dependencies with:

uv add openai httpx
uv sync

Core dependencies you'll always need:

• agentex-sdk - Required for all agents

---

Dockerfile

Defines how your agent is packaged into a container. It starts with a Python base image, installs system and Python dependencies, copies your code, and runs either the ACP server or worker.

You don't often need to change this unless you know what you're doing and need to add custom files, folders, or system packages directly.

FROM python:3.12-slim
COPY --from=ghcr.io/astral-sh/uv:0.6.4 /uv /uvx /bin/

# Install system dependencies
RUN apt-get update && apt-get install -y \
    build-essential \
    postgresql-client \
    && apt-get clean

# Install Python dependencies
COPY my-agent/pyproject.toml /app/my-agent/pyproject.toml
WORKDIR /app/my-agent
RUN uv pip install --system .

# Copy agent code
COPY my-agent/project /app/my-agent/project

# Run ACP server
CMD ["uvicorn", "project.acp:acp", "--host", "0.0.0.0", "--port", "8000"]

For Temporal agents deployed as workers, the CMD is replaced with:

CMD ["python", "-m", "project.run_worker"]

---

.dockerignore

Excludes unnecessary files from the Docker build context. This keeps builds fast and prevents secrets from leaking into images.

# Python
__pycache__/
*.pyc
.venv/

# Environment
.env
.env.local

# Git
.git/
.gitignore

# IDE
.idea/
.vscode/

# Misc
.DS_Store

---

dev.ipynb

An interactive Jupyter notebook for testing your agent locally. It includes cells for connecting to your local backend, creating tasks, sending messages/events, and subscribing to responses.

# Setup
from agentex import AsyncAgentex
client = AsyncAgentex(base_url="http://localhost:8080")

# Create task
agent = await client.agents.retrieve(name="my-agent")
task = await client.tasks.create(agent_id=agent.id)

# Send message (Sync agents)
response = await client.messages.send(
    task_id=task.id,
    agent_id=agent.id,
    content=TextContent(author="user", content="Hello!")
)

# Send event (Async/Temporal agents)
await client.events.send(
    task_id=task.id,
    agent_id=agent.id,
    content=TextContent(author="user", content="Hello!")
)

# Subscribe to responses
async for message in client.tasks.subscribe(task_id=task.id):
    print(f"{message.content.author}: {message.content.content}")

---

.env (Optional)

Store local environment variables for development. This file is automatically loaded when placed alongside your manifest.yaml file, but only for local development.

# .env
OPENAI_API_KEY=sk-your-key-here
DATABASE_URL=postgresql://localhost/mydb
CUSTOM_API_URL=https://api.example.com

For production: Use the credentials section in your manifest.yaml instead. See the Agent Configuration Files documentation for more details.

Security

Never commit .env to git! It's automatically excluded by .dockerignore and should be in your .gitignore.