coreloop

A lightweight async tool-calling agent for any OpenAI-compatible API (via httpx). The core is an async generator loop that streams Message objects; you observe and intercept it via lifecycle hooks. Usable as a library or through a minimal CLI.

The minimal core imposes no forced overhead, so it pairs well with small local models; early exit via hooks and stop() makes high-throughput batch work fast and keeps token costs down.

Built-in tools: path-scoped read, ls, edit, grep; a bash tool with best-effort guardrails (not a security sandbox — see below); optional web_search and web_fetch (via the [web] extra).

• Observability — hook into every stage of the loop: before/after each turn, LLM call, and tool execution — examples: streaming · raw message stream · logging & timing hooks
• Loop control — stop cleanly, abort immediately, or inject responses mid-run: stop(), abort(), on_before_llm — examples: lifecycle control · intercept & replace results
• Extensibility — register custom async tools with @tool: inferred JSON Schema, callable directly or by name — examples: custom tools · subagents

Install

Requires Python 3.12+.

pip install coreloop
pip install "coreloop[web]"   # adds web_search and web_fetch

Library quick-start

import asyncio
from coreloop import Agent, Message

agent = Agent(
    model="gpt-4o-mini",
    base_url="https://api.openai.com/v1",
    api_key="sk-...",
    tools=["read", "ls", "grep"],
    root="/tmp/sandbox",
)

async def main():
    async for msg in agent.run([Message(role="user", content="What files are here?")]):
        if msg.role == "assistant" and not msg.partial and msg.content:
            print(msg.content)

asyncio.run(main())

Or load settings from a named profile in ~/.coreloop.toml:

agent = Agent.from_profile("openai")

CLI

core is a REPL / one-shot runner with profile support. On first run it copies the bundled coreloop.toml to ~/.coreloop.toml — edit that file to set your default model, tools, and provider credentials.

# Interactive REPL using the default profile (Ollama)
core

# One-shot with a named profile
core --profile openai -p "Summarise this repo"

# Override model and enable file tools
core --profile openai --model gpt-4o --tools read,ls,grep --root .

# Bypass profiles entirely
core --base-url https://api.openai.com/v1 --api-key $OPENAI_API_KEY --model gpt-4o-mini -p "Hello"

# Enable reasoning
core --think -p "Explain this step by step" --model qwen3-14b

REPL commands: /quit /exit /q to exit; /new to clear history; /model <name> to switch models; /root <path> to change the file-tool root.

Agent

Agent(
    model: str,
    base_url: str = "http://localhost:11434/v1",
    api_key: str | None = None,
    system: str | None = None,
    tools: list[str | ToolInfo] | None = None,
    root: str | Path | None = None,
    http_request_timeout: float = 300.0,  # httpx per-chunk read timeout
    tool_timeout: float = 360.0,          # hard wall per tool call
    llm_timeout: float = 300.0,           # asyncio wall for the entire LLM turn
    hooks: AgentHooks | None = None,
    llm_extra_body: dict | None = None,
    cache_dir: Path | str | None = "~/.cache/coreloop-llm-cache",
)

tools accepts built-in names ("read", "ls", "edit", "grep", "bash", "web_search", "web_fetch"), names of @tool-registered functions, or ToolInfo objects. File tools are scoped to root; an unknown name raises ValueError. An agent has exactly the tools you list — there is no implicit inclusion of the global registry.

run(messages) is an async generator. Partial streaming chunks have partial=True; the final assembled message for each LLM turn has partial=False. Pass usage=Usage() to accumulate token counts across turns.

Restart pattern — pass agent.messages to keep history across runs:

async for msg in agent.run([Message(role="user", content="Hello")]):
    ...

agent.model = "stronger-model"
async for msg in agent.run(agent.messages + [Message(role="user", content="Now do X")]):
    ...

Built-in tools

All file tools reject path traversal and are scoped to root.

bash is not sandboxed. It runs whatever the model sends with your full user privileges. The workdir is scoped to root, but a command can still read, write, or delete anything your account can reach, and make network calls. The dangerous-pattern blocklist (e.g. rm -rf /, mkfs, fork bombs) is a speed bump against obvious accidents, not a security boundary — trivial variants slip through (curl … | bash, python -c "…", unusual flag orders). Only enable bash for models and prompts you trust, and prefer running in a container or VM.

Web tools require pip install "coreloop[web]" and a running SearXNG instance:

docker run -d -p 8080:8080 searxng/searxng

Custom tools

from coreloop import tool

@tool
async def read_env(name: str) -> str:
    """Read an environment variable."""
    import os
    return os.environ.get(name, "(not set)")

# Attach by name or pass the ToolInfo object directly
agent = Agent(model="...", tools=["read_env"])
agent = Agent(model="...", tools=[read_env])  # equivalent

@tool infers JSON Schema from type annotations (str, int, float, bool, list[T], dict, Optional[T]). Unrecognised types fall back to string with a warning. Override name or description:

@tool(name="my_read", description="Read a local file")
async def _impl(path: str) -> str:
    ...

@tool(allow_override=True)   # re-register if a tool with that name already exists
async def read_env(name: str) -> str:
    ...

The decorated object is both a ToolInfo and directly callable (await my_tool(...) works alongside registration).

Registry helpers:

from coreloop import list_tools, get_tool, clear_registry

list_tools()          # list[ToolInfo] — all globally registered tools
get_tool("read_env")  # ToolInfo | None
clear_registry()      # remove all tools (useful in tests)

Hooks

Subclass AgentHooks to observe or intercept any stage of the loop:

from coreloop import AgentHooks

class LogHook(AgentHooks):
    async def on_before_tool(self, agent, name, args):
        print(f"→ {name}({args})")
        return None  # None = run the tool; return str to inject a result instead

agent = Agent(..., hooks=LogHook())

Hook firing order per turn:

on_before_turn
on_before_llm   → return Message to skip the LLM call entirely
  <LLM streams>
on_after_llm    → return Message to replace before appending to history
  for each tool (in parallel):
    on_before_tool  → return str to skip tool execution
      <tool runs>
    on_after_tool   → return str to replace result in history
on_after_turn

on_before_agent / on_after_agent bracket the entire run() call. on_after_agent is not called after abort(). Hook exceptions are logged and swallowed — they cannot crash the agent.

MaxTurnsHook(n) calls agent.stop() after n turns; the counter resets on each run(), so it enforces a per-run limit, not a lifetime one.

Config profiles

Agent.from_profile("openai") builds an AgentConfig from ~/.coreloop.toml (or $CORELOOP_CONFIG). Every profile inherits from [profiles.default]; named profiles override individual keys.

[profiles.default]
base_url = "http://localhost:11434/v1"
model    = "qwen3.5:9b"

[profiles.openai]
base_url = "https://api.openai.com/v1"
api_key  = "{{OPENAI_API_KEY}}"
model    = "gpt-4o-mini"
tools    = ["read", "ls", "grep"]

[profiles.together]
base_url = "https://api.together.xyz/v1"
api_key  = "{{TOGETHER_API_KEY}}"

{{VAR_NAME}} in any string value is interpolated from the environment. Unknown keys are silently ignored. The shipped src/coreloop/coreloop.toml includes pre-configured profiles for Ollama, OpenAI, Groq, DeepSeek, Together, and OpenRouter.

AgentConfig is a dataclass mirroring the Agent constructor (excluding hooks). Use dataclasses.replace(cfg, model="other") to derive variants, or Agent.from_config(cfg) to construct an agent from one.

Structured output

complete() is a one-shot call — no tools, no loop. With schema, output is constrained via response_format: {"type": "json_schema"} and validated before it is returned.

from pydantic import BaseModel
from coreloop import complete

class Triage(BaseModel):
    category: str
    urgent: bool

result = await complete("Kettle broke again.", model="qwen3:8b", schema=Triage)
# -> Triage(category='...', urgent=True)

Without schema it returns the final Message. ValidationError propagates to the caller, so retry policy stays yours.

prompt also accepts a message list, which makes "loop, then extract" compose:

async for _ in agent.run([Message(role="user", content="Check the logs and triage.")]):
    pass

record = await complete(agent.messages, model="qwen3:8b", schema=Triage)

It is deliberately not an Agent parameter: constrained decoding and tool-calling cannot coexist in one turn, so the loop stays unconstrained and complete() is the terminal extraction step.

Message

Caching

LLM responses are disk-cached by default (~/.cache/coreloop-llm-cache/), keyed by SHA-256 of (model, messages, tools, extra_body). This replays identical requests without hitting the API — useful during development. Pass cache_dir=None to disable.

agent = Agent(model="...", cache_dir="/tmp/my-cache")  # custom location
agent = Agent(model="...", cache_dir=None)             # disabled