abi-core-ai 1.13.6

Agent-Based Infrastructure Core - Runtime and CLI

ABI-Core AI 🤖

Build AI agents that work together, find each other, and follow the rules.

ABI-Core is a Python framework for creating AI agents. You write the logic as simple functions, ABI packages them into services, connects them to each other, and makes sure they play by the rules. One pip install, one CLI command, and you have a running agent system.

pip install abi-core-ai
abi-core create swarm --name my-system  # beta
abi-core run

⚠️ Beta — Pipeline works end-to-end. APIs may change between minor versions.

🔄 v1.12+: Requires a2a-sdk>=1.0.0. If you have an existing project with a custom config.py using AgentCard(**data), see the migration guide. Projects running on the ABI Docker image are unaffected.

---

Create an Agent in 3 Files

1. Define the steps (app.py)

from abi_core.agent import AbiCore
from .my_agent import MyAgent

agent = AbiCore()

@agent.step(name="parse_input")
async def parse_input(raw_input: str) -> dict:
    return {"query": raw_input.strip(), "timestamp": time.time()}

@agent.step(name="process", depends_on=["parse_input"], input_map={"data": "$parse_input.result"})
async def process(data: dict) -> dict:
    result = await invoke(config.LLM_CONFIG, data["query"])
    return {"output": result}

@agent.step(name="respond", depends_on=["process"], input_map={"result": "$process.result"})
async def respond(result: dict) -> dict:
    return {"response": result["output"]}

agent.run(MyAgent())

2. Define the agent (my_agent.py)

from abi_core.agent import AbiAgent

class MyAgent(AbiAgent):
    def __init__(self):
        super().__init__(
            agent_name="my-agent",
            description="Processes user queries",
            llm_config={"provider": "ollama", "model": "qwen3:8b", "temperature": 0.3},
            system_prompt="You are a helpful assistant.",
        )

3. Configure it (config/config.py)

import os

AGENT_NAME = "my-agent"
DESCRIPTION = "Processes user queries"
LLM_CONFIG = {"provider": "ollama", "model": "qwen3:8b", "temperature": 0.3}
OLLAMA_HOST = os.getenv("OLLAMA_HOST", "http://localhost:11434")

That's it. abi-core run packages and starts your agent with messaging, health checks, and automatic registration.

---

Key Concepts

Decorators

Decorator	What it does
@agent.step(name, depends_on)	A function that runs in a fixed order you define
@agent.tool(name)	A function the AI can decide to call
@agent.mcp_tool(name)	A remote tool on the Semantic Layer
@agent.task(name, task_id)	Runs steps in sequence with progress updates

Steps run in order

Steps run in the order you define with depends_on. Steps at the same level run in parallel. The AI never decides execution order — your code does.

# These two run at the same time (no dependency between them)
@agent.step(name="classify")
async def classify(raw_input): ...

@agent.step(name="validate")
async def validate(raw_input): ...

# This waits for both to finish
@agent.step(name="decide", depends_on=["classify", "validate"],
            input_map={"cls": "$classify.result", "valid": "$validate.result"})
async def decide(cls, valid): ...

invoke() — Call any AI model

from abi_core.agent import invoke

# Simple call
result = await invoke(config.LLM_CONFIG, "Classify this query...")

# With conversation memory
result = await invoke(config.LLM_CONFIG, "Follow up...", thread_id=session_id)

# With tools the AI can use
result = await invoke(config.LLM_CONFIG, "Find...", tools=[search_tool, write_tool])

Memory — short & long-term

Give agents system-wide memory backed by the Agent Memory Server. Store deliberately, recall on demand — across steps, tasks, and sessions:

from abi_core.agent import (
    add_short_term_memory, add_long_term_memory,
    get_long_term_memory, recall_memory_context,
)

# Write (inside a step/task)
await add_short_term_memory("processing", "pipeline", "processed 42 records", context_id=ctx)
await add_long_term_memory("user_preference", "format", "prefers CSV over PDF", context_id=ctx)

# Read
past = await get_long_term_memory("format preferences")
context = await recall_memory_context(query, context_id=ctx)  # ready to inject into a prompt

Every call degrades gracefully when the memory server is unavailable — memory never blocks execution. See the memory guide.

Sessions — reliable multi-turn

Opt-in sessions tie an opaque, backend-generated token to an internal context_id and its conversation context. Pick the backend with one env var — your agent code doesn't change:

from abi_core.agent import SessionStore

store = SessionStore.from_env()              # SESSION_BACKEND=memory | redis
session = await store.create_session()        # opaque token + backend context_id
ctx = await store.get_context(session.context_id)

Use SESSION_BACKEND=redis for multi-pod / load-balanced deployments: state lives in shared Redis, not per-process RAM, so a follow-up request survives a pod hop. See the sessions guide.

Capability matching — pick models by what the task needs

Instead of "which model should we use?", ask "what capabilities does the task require?". Profile a task by what it needs and a model by what it provides, over 7 dimensions (including instruction_following — why a small model can be the better orchestrator), and match them:

from abi_core.capabilities import TaskProfile, CapabilityProfile, select_model, seed_catalog

task = TaskProfile(capabilities=CapabilityProfile(code_generation=0.9, tool_usage=0.95))
result = select_model(task, seed_catalog())
result.model_name   # best match; result.gaps → deficits to reinforce

Measure your own models with a deterministic probe battery (Wilson confidence intervals, no LLM judge) and export to JSON:

abi-core capabilities profile qwen2.5:3b --output profiles.json
abi-core capabilities show qwen2.5:3b --source profiles.json   # radar in the terminal

See the capability matching guide.

Agents talk to each other

from abi_core.common.abi_a2a import agent_connection

async for chunk in agent_connection(my_card, target_card, payload):
    process(chunk)

---

CLI

# Create
abi-core create swarm --name <name>          # Full system: agents + services + compose (beta)
abi-core create project <name>               # Project only
abi-core add agent <name> --description "…"  # Add agent to existing project
abi-core add semantic-layer                  # Add agent discovery service
abi-core add service guardian-native         # Add security gate
abi-core add chainlit                        # Add a Chainlit chat UI as a Docker service (SSE + sessions)

# Run
abi-core run                # Start everything
abi-core run --logs         # With container output
abi-core run --build        # Rebuild first

---

Built-in Agents

When you create swarm, you get these out of the box:

Agent	What it does
Orchestrator	Receives requests, checks security, routes to Planner
Planner	Breaks complex requests into smaller tasks
Builder	Creates temporary agents on-demand for specific tasks (beta)
Zombie	Temporary agent — does the work, delivers results, cleans up (beta)

The flow: User → Orchestrator → Planner → Builder → Zombie → Result → Done.

---

Any AI Model

Switch providers by changing one config dict. Same code, any model:

# Local (Ollama)
{"provider": "ollama", "model": "qwen3:8b"}

# OpenAI
{"provider": "openai", "model": "gpt-4o", "api_key": "..."}

# Anthropic
{"provider": "anthropic", "model": "claude-sonnet-4-20250514"}

# AWS Bedrock
{"provider": "bedrock", "model": "anthropic.claude-3-sonnet"}

# Azure OpenAI
{"provider": "azure", "model": "gpt-4o", "endpoint": "..."}

---

Security

• Guardian — checks every request against rules before it runs
• Signed messages — agent-to-agent calls are signed and verified
• Access control — agents can only use tools they're allowed to
• Audit trail — every decision is logged with a risk score
• Human veto — you can block execution before it starts

---

Project Structure

my-swarm/
├── agents/
│   ├── orchestrator/     # Receives and routes requests
│   ├── planner/          # Breaks tasks into pieces
│   ├── builder/          # Creates temporary agents
│   └── my-agent/         # Your custom agents
├── services/
│   ├── semantic_layer/   # Agent discovery + search
│   └── guardian/         # Security rules
├── compose.yaml
└── .abi/runtime.yaml

---

Examples

Progressive examples from a simple chatbot to a full multi-agent swarm:

👉 abi-core-examples — Includes a step-by-step tutorial for building a multi-agent discussion system.

---

Documentation

Full docs: https://abi-core.readthedocs.io

---

Contributing

git clone https://github.com/Joselo-zn/abi-core
cd abi-core-ai
uv sync --dev
uv run pytest

---

License

Apache 2.0 — see LICENSE

---

Built by José Luis Martínez