Skip to main content

Your First Agent

This guide walks you through creating and running a minimal custom agent using the agenticstar-platform SDK. The SDK provides infrastructure (DB, events, storage, etc.), while you freely design the agent logic.

Prerequisites​

  • Python 3.12 or higher
  • pip or uv
  • PostgreSQL database (local or Azure Database for PostgreSQL)
  • Azure OpenAI API key (for LLM calls)

Step 1: Install the SDK​

curl
pip install agenticstar-platform[all]==3.0.15

Step 2: Create the Configuration File​

Create a config.toml file. Each SDK module reads its configuration from this file.

config.tomlToml
[database]
host = "localhost"
port = 5432
database = "agenticstar"
username = "agent_user"
password = "your_password"
pool_min_size = 2
pool_max_size = 10
# Set to true only when using Azure AD auth, and configure [database.azure_ad]
use_azure_ad = false

[events]
enable_db_handler = true
enable_webhook_handler = false

PostgreSQLConfig.from_toml() reads the [database] section by default. The field names are username / pool_min_size / pool_max_size. To use Azure AD auth, set use_azure_ad = true directly under [database] and provide tenant_id / client_id / client_secret in [database.azure_ad].

Step 3: Implement the Agent​

Create my_agent.py. This example implements a simple agent that retrieves job information using the SDK's DB module and streams progress via the Events module.

my_agent.pyPython
import asyncio
from openai import AsyncAzureOpenAI

from agenticstar_platform.db import PostgreSQLManager, PostgreSQLConfig, DataAccess
from agenticstar_platform.db.execution_access import ExecutionAccess
from agenticstar_platform.events import EventEmitter, EventType

# --- Configuration ---
db_config = PostgreSQLConfig.from_toml("config.toml")
llm_client = AsyncAzureOpenAI(
    azure_endpoint="https://your-endpoint.openai.azure.com/",
    api_version="2024-12-01-preview",
    api_key="your-api-key",
)

async def run_agent(execution_id: str, user_message: str):
    """Minimal agent execution flow"""

    # 1. Initialize DB connection
    db_manager = PostgreSQLManager(db_config)
    da = DataAccess(db_manager)
    await da.initialize()

    execution_access = ExecutionAccess(da)

    # 2. Initialize EventEmitter
    emitter = EventEmitter(execution_id=execution_id)

    # 3. Emit execution start event
    await emitter.emit_event(
        event_type=EventType.PHASE_START,
        message="Processing request...",
        metadata={"phase": "processing"},
    )

    # 4. Retrieve past messages
    history = await execution_access.get_messages(execution_id=execution_id)

    # 5. Call LLM (agent logic is freely designed)
    messages = [{"role": "system", "content": "You are a helpful assistant."}]
    if history:
        for h in history:
            messages.append({"role": h["role"], "content": h["content"]})
    messages.append({"role": "user", "content": user_message})

    response = await llm_client.chat.completions.create(
        model="gpt-4.1",
        messages=messages,
    )
    answer = response.choices[0].message.content

    # 6. Emit completion event
    await emitter.emit_event(
        event_type=EventType.COMPLETION_SUCCESS,
        message=answer,
    )

    # 7. Cleanup
    await emitter.cleanup()
    await da.close()

    return answer

if __name__ == "__main__":
    result = asyncio.run(run_agent("exec-001", "Hello!"))
    print(result)

Step 4: Verify​

curl
python my_agent.py

If everything works correctly, the LLM response will be printed to the console.

Run Marketplace-Compatible (runner, SDK 0.5.30+)​

An agent function that works locally can run under the marketplace-compatible terminal lifecycle just by passing it to run_marketplace_agent. The runner takes care of receiving and validating identity (EXECUTION_ID etc.), fetching the input message, persisting the result to the DB, webhook notification, the terminal event (exactly once, no matter what happens), and cleanup — so the main boilerplate you wrote in Step 3 above is no longer needed.

marketplace_agent.pyPython
from agenticstar_platform import run_marketplace_agent


async def my_agent(emitter, message: str) -> str:
    """Your agent logic. Any framework (LangChain / OpenAI Agents SDK / custom)."""
    return message.upper()  # replace with your actual logic


if __name__ == "__main__":
    run_marketplace_agent(my_agent)
curl
pip install 'agenticstar-platform[runner]==3.0.15'
python marketplace_agent.py

On the platform, EXECUTION_ID / CONVERSATION_ID / USER_ID / MESSAGE_ID are injected by the marketplace executor at pod startup. Configure the DB connection (DB_HOST etc., following the PostgreSQLConfig.from_env() contract) and WEBHOOK_URL as agent environment variables at registration time. If a required variable is missing, the runner stops with MarketplaceRunnerConfigError without invoking your agent.

  • The agent function's return value is persisted and notified as completion_success
  • Exceptions from the agent function converge to completion_failure (traceback goes to logs only)
  • Agent functions that emit their own terminal event (like the local sample) do not cause duplicates
  • For cycle executions (mp-chat/1, conversations started by a project cycle on a schedule or by a webhook), PROJECT_ID (the ID of that project) is injected in addition to the variables above. For each enabled connection of the project, the OAuth token is passed as ASTER_CONNECTION_<SERVICE>_TOKEN (<SERVICE> is the service name of the connection in upper case, with - replaced by _; for example, slack_mcp → ASTER_CONNECTION_SLACK_MCP_TOKEN). The execution does not start when more than 8 connections are enabled, when two connections of the same service are enabled, or when this variable name collides with an environment variable from the agent registration or the placement. Tokens of connections that are disconnected, or expired and cannot be refreshed, are not passed (an expired token is passed if it can be refreshed), so make your agent handle the case where the variable is missing. Your agent looks up the connection URLs from the settings itself (for example, get_mcp_configurations(execution_mode="pod"))
  • For cycle executions (mp-chat/1, conversations the platform starts on a schedule), from v3.0.12 the runner takes a started fence in the database before calling your agent. If the pod is recreated and the same execution starts again, your agent is not called and the runner raises MarketplaceRunnerFenceTaken (the platform closes the execution as failed with unknown effects). The fence uses the direct PostgreSQL pool (PostgreSQLManager.pool); with a data_access that has no pool, pass started_fence=False (a recreated pod may then run your agent again)

Exit the process after a cycle execution​

Required if you run your own server

A cycle execution (mp-chat/1) runs in a Pod created for that single execution (restartPolicy: Never). The platform decides that an execution has finished by looking at the container exit code, so if the process stays alive after your work is done, the execution never finishes.

While it never finishes, the next scheduled trigger for that placement keeps being skipped (the previous execution is still considered running).

Cycle executions use the same protocol as chat, so this happens if you reuse an example that keeps a server answering health checks.

If you pass your agent to run_marketplace_agent (above), the runner handles termination and you need no extra work.

If you run your own server (FastAPI / uvicorn and the like), close the server and exit after the work completes. Keep answering health checks while the work is in progress, so close it only after you have confirmed completion.

start_api_server.py (from the sample agent)Python
import asyncio, os
import uvicorn

# Set at the end of your agent body to signal that the run is done
_run_finished = asyncio.Event()


def exit_after_run() -> bool:
    """A cycle execution runs once, so end the process when the run completes."""
    return os.environ.get("AGENT_MODE", "").strip().lower() == "cycle"


async def serve(config: uvicorn.Config) -> None:
    server = uvicorn.Server(config)
    serving = asyncio.create_task(server.serve())
    if exit_after_run():
        finished = asyncio.create_task(_run_finished.wait())
        # Do not wait for completion forever if the server ends first (startup failure, SIGTERM)
        await asyncio.wait({serving, finished}, return_when=asyncio.FIRST_COMPLETED)
        finished.cancel()
        # Closing before startup finishes cancels the cleanup, so wait for `started`
        for _ in range(100):  # up to 5 seconds
            if server.started or serving.done():
                break
            await asyncio.sleep(0.05)
        server.should_exit = True
    await serving


if __name__ == "__main__":
    asyncio.run(serve(uvicorn.Config("main:app", host="0.0.0.0", port=int(os.environ.get("PORT", "8000")))))
  • Switch on cycle mode with an environment variable set in the agent registration (for example AGENT_MODE=cycle). If the same image also serves chat, keep the server resident for chat
  • Do not call the uvicorn APIs that prepare the loop (Config.setup_event_loop / Config.get_loop_factory). They differ between versions, and setup_event_loop was removed in 0.36.0
  • Exit with code 0 even when the work fails (the runner does this for you). Report what went wrong through completion reporting

Reporting completion for cycle executions (v3.0.14+)​

A cycle execution whose agent finishes normally is recorded as "finished, completion not measured" and is not counted as a success, because a clean process exit is not treated as completion. To have it counted, your agent reports whether it achieved its goal with report_completion. Reporting is optional; agents that do not report behave as before.

marketplace_agent.py (reporting completion)Python
import asyncio
import os

from agenticstar_platform import arun_marketplace_agent, report_completion
from agenticstar_platform.db import DataAccess, PostgreSQLConfig, PostgreSQLManager


async def main() -> None:
    # build the DataAccess yourself and pass it to the runner, so that you can use its pool
    manager = PostgreSQLManager(PostgreSQLConfig.from_env())
    async with DataAccess(manager) as data_access:

        async def my_agent(emitter, message: str) -> str:
            reply = message.upper()  # replace with your actual processing
            await report_completion(
                manager.pool,
                user_id=os.environ["USER_ID"],
                conversation_id=os.environ["CONVERSATION_ID"],
                message_id=os.environ["MESSAGE_ID"],
                execution_id=os.environ["EXECUTION_ID"],
                outcome="completed",          # e.g. "incomplete" if the goal was not achieved
                had_mutating_success=False,   # wrote something outside? None if unknown
            )
            return reply

        await arun_marketplace_agent(my_agent, data_access=data_access)


if __name__ == "__main__":
    asyncio.run(main())
  • Choose outcome from completed / incomplete / failed / timeout / stopped / denied / skipped. What counts as completion is up to your agent
  • To give a reason, pass a short snake_case code as reason (for example no_new_items) and put sentences in summary. The platform uses only outcome and reason for the execution's result
  • Report only once (a second call, even a concurrent one, writes nothing and returns False). A failed database write does not raise; the execution just stays unmeasured
  • The platform adopts the report only when it sees the container exit 0 (not after a cancellation, the deadline passing while your agent runs, or a lost pod). The runner exits 0 even when your agent raises, so a report already written is adopted in that case too. Report once the result is final (at the end of your agent function)
  • Calling it in a normal (chat) execution changes nothing. For the arguments and behavior, see the SDK API Reference

Local Integration Lab (SDK 0.5.35+)​

Experience the full integration journey across DB / RAG / storage (ingest → retrieve → artifact/result persist → terminal outcome) with a single command — no external service contracts, no cloud credentials, no .env editing. All you need is Docker (with compose v2) and Python 3.11+.

curl
pip install 'agenticstar-platform[lab]==3.0.15'
python -m agenticstar_platform.lab            # run the journey (one command)
python -m agenticstar_platform.lab doctor     # read-only diagnostics
python -m agenticstar_platform.lab --reset    # tear down (lists targets, confirms / --yes)
  • Starts PostgreSQL / Qdrant / S3-compatible storage (MinIO) via a version-pinned Docker Compose manifest (bound to 127.0.0.1 only)
  • Embeddings use a deterministic offline implementation — no cloud endpoint or API key is ever used
  • The only credentials involved are fixed, non-production lab values; the lab never asks for, stores, or prints production credentials
  • A local copy of the artifact is written to agenticstar-lab-artifacts/ under the current working directory
  • On success it shows the retrieved document, the artifact location, and the next Marketplace runner command to try. If something fails, doctor classifies the cause (Docker not running / port conflict / missing extras etc.) with the next command to run

Step 5: Add Tools (Optional)​

Here is an example of adding tools (function calling) to your agent. This uses the SDK's RAG module to search a knowledge base.

tools/search_knowledge.pyPython
from agenticstar_platform.rag import QdrantManager, QdrantConfig
from agenticstar_platform.rag import EmbeddingGenerator, EmbeddingConfig

async def search_knowledge(query: str, top_k: int = 5) -> list[dict]:
    """Search the knowledge base for documents related to the query"""

    # Initialize the embedding generator and the Qdrant manager
    # QdrantManager requires an EmbeddingGenerator (it embeds the query internally at search time)
    embed_config = EmbeddingConfig.from_toml("config.toml")
    generator = EmbeddingGenerator(embed_config)

    qdrant_config = QdrantConfig.from_toml("config.toml")  # collection_name comes from config
    qdrant = QdrantManager(qdrant_config, generator)
    await qdrant.initialize()

    # Search by query text (no query_vector needed; embedding is generated internally)
    result = await qdrant.search(query_text=query, limit=top_k)

    # The return value is a dict; hits are in result["data"]["results"]
    return [
        {"content": hit["payload"].get("content", ""), "score": hit["score"]}
        for hit in result["data"]["results"]
    ]

Example Project Structure​

my-agent/
├── config.toml              # SDK configuration file
├── my_agent.py              # Agent main logic
├── tools/
│   ├── search_knowledge.py  # RAG search tool
│   └── file_manager.py      # Storage operations tool
├── Dockerfile               # Container build
├── requirements.txt
└── README.md

Next Steps​