neuroagent-ai 0.1.1

The FastAPI of AI Agents — build production-ready AI agents, teams, RAG, and workflows from one package.

NeuroAgent AI

The FastAPI of AI Agents. One framework. Infinite agents.

Build production-ready AI agents, multi-agent teams, RAG systems, database & API agents, and autonomous workflows — from a single, provider-agnostic package with minimal boilerplate.

from neuroagent import Agent

agent = Agent(provider="anthropic", model="claude-opus-4-8")
print(agent.run("Analyze our Q3 sales data and write a short summary."))

---

Table of contents

• Why NeuroAgent AI
• Install
• Quickstart (60 seconds, no API key)
• How it works
  • Architecture
  • The agent run loop
  • Design principles
• Configuration
• Core concepts & use cases
  • 1. Agents
  • 2. Providers
  • 3. Tools
  • 4. Memory
  • 5. RAG (chat with your documents)
  • 6. Database agents (NL → SQL)
  • 7. API agents
  • 8. Workflows (DAG orchestration)
  • 9. Multi-agent teams
  • 10. Streaming
  • 11. Observability & cost
  • 12. Security (RBAC, audit, secrets, sandbox)
  • 13. Business integrations
  • 14. Deploy as an HTTP server
• Putting it together
• Documentation
• Contributing & releasing
• License

---

Why NeuroAgent AI

Building a reliable AI agent today means stitching together LLM SDKs, prompt systems, tool calling, memory, vector stores, retrieval, databases, orchestration, observability, and cost tracking. Every project rebuilds the same plumbing, and switching model providers means rewriting it.

NeuroAgent AI puts all of that behind one clean, provider-agnostic API:

You get	Instead of
One Agent class	Gluing 5+ SDKs together
provider="openai" | "anthropic" | "gemini" | "groq"	Vendor lock-in
memory="sqlite", rag=True, tools=[...]	Custom infra per project
Workflow, Team, DatabaseAgent, APIAgent	Reinventing orchestration
agent.stats(), traces, agent.serve()	Bolt-on observability & servers

Everything is typed (mypy --strict), tested (~260 tests, offline), and lean — the base install has no vendor SDKs; heavy/optional features live behind extras.

---

Install

pip install neuroagent-ai                 # core + all LLM providers (httpx-based, no vendor SDKs)

pip install "neuroagent-ai[rag]"          # + document loaders (PDF/DOCX) for retrieval
pip install "neuroagent-ai[sql]"          # + SQLAlchemy for Postgres/MySQL/etc. database agents
pip install "neuroagent-ai[redis]"        # + Redis memory backend
pip install "neuroagent-ai[security]"     # + secret encryption (cryptography/Fernet)
pip install "neuroagent-ai[server]"       # + FastAPI/uvicorn for agent.serve()
pip install "neuroagent-ai[all]"          # everything

Lean by design. OpenAI, Anthropic, Gemini, and Groq all work on the base install (they talk to the REST APIs over httpx). Extras are only for genuinely heavier capabilities. Missing an extra raises a clear MissingDependencyError telling you exactly what to pip install.

Requires Python 3.10+.

---

Quickstart (60 seconds, no API key)

Every example below runs offline with the built-in echo provider (it echoes your prompt — no key, no network), so you can explore the API instantly. Swap provider="echo" for a real provider + API key when you're ready.

from neuroagent import Agent, tool

@tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

agent = Agent(provider="echo", tools=[add])
response = agent.run("Please add 2 and 3")

print(response.content)              # the answer text
print(response.usage.total_tokens)   # token accounting
print(response.steps)                # provider/tool iterations

To use a real model, set the provider and an API key:

import os
os.environ["OPENAI_API_KEY"] = "sk-..."

agent = Agent(provider="openai", model="gpt-4.1", tools=[add])
print(agent.run("What is 2 + 3? Use the tool.").content)

---

How it works

Architecture

                          ┌─────────────────────────────┐
        Your code  ─────► │            Agent            │   (public API, sync + async)
                          └──────────────┬──────────────┘
                                         │ orchestrates
        ┌───────────┬───────────┬────────┼────────┬───────────┬───────────┐
        ▼           ▼           ▼        ▼        ▼           ▼           ▼
    Providers     Tools      Memory     RAG    Context     Cost       Tracing
   (LLM calls) (functions) (history) (retrieval)(prompt)  (tokens→$)  (spans)

  Higher-order compositions, all built ON TOP of Agent (never around it):
     Workflow ── DAG of steps       Team ── collaborating agents      Server ── agent.serve()
     DatabaseAgent ── NL→SQL        APIAgent ── call HTTP APIs        Integrations ── Shopify/SAP/…

The Agent is the atom. Teams, workflows, database agents, and API agents are compositions or specializations of Agent, not parallel systems. Learn Agent once and everything else follows.

The agent run loop

When you call agent.run(prompt) (sync) or await agent.arun(prompt) (async):

1. Assemble context  →  system prompt + memory history + (optional) retrieved RAG context + prompt
2. Call the provider →  normalized LLM request (with your tools' JSON schemas)
3. Tool calls?       →  validate args, execute the tool, feed the result back ──┐
                        (loop, bounded by max_steps)  ◄────────────────────────┘
4. Final answer      →  return AgentResponse(content, messages, usage, steps, trace_id)
5. Along the way     →  record a span tree (trace) + token usage → cost; persist to memory

Tools auto-generate their JSON schema from your function's signature, type hints, and docstring — you never hand-author schemas. Every run is traced and costed (see Observability).

Design principles

• Provider-agnostic — all models go through one LLMProvider contract; no core code depends on a vendor.
• Async-first, sync-friendly — every async method (arun, astream, …) has a sync twin (run, …).
• Safe by default — database & API agents are read-only unless you opt in to writes.
• Errors are UX — every exception derives from NeuroAgentError, names the subsystem, and tells you the fix.
• Observable — provider calls, tool calls, memory, and retrieval all emit spans + normalized token usage.

---

Configuration

Pass options directly, or set environment variables. Precedence: explicit kwargs > env vars > defaults.

from neuroagent import Agent, Settings

# Per-agent
agent = Agent(
    provider="anthropic",
    model="claude-opus-4-8",
    api_key="sk-ant-...",       # or rely on the env var below
    system_prompt="You are a concise financial analyst.",
    temperature=0.2,
    max_tokens=1024,
    max_steps=8,                # cap on tool-calling iterations
)

# Global defaults via env (prefix NEUROAGENT_) or a .env file
#   NEUROAGENT_DEFAULT_PROVIDER=anthropic
#   NEUROAGENT_DEFAULT_MODEL=claude-opus-4-8
settings = Settings(default_provider="openai", default_model="gpt-4.1")
agent = Agent(settings=settings)   # provider/model resolved from settings

Provider API keys are read from the standard environment variables: OPENAI_API_KEY, ANTHROPIC_API_KEY, GEMINI_API_KEY (or GOOGLE_API_KEY), GROQ_API_KEY.

---

Core concepts & use cases

1. Agents

from neuroagent import Agent

agent = Agent(provider="openai", model="gpt-4.1", system_prompt="Answer in one sentence.")

resp = agent.run("Who wrote Pride and Prejudice?")
print(resp.content)          # final text
print(resp.usage)            # Usage(prompt_tokens=..., completion_tokens=...)
print(resp.steps)            # number of model/tool iterations
print(resp.trace_id)         # links to this run's trace

# Async + streaming variants
import asyncio
asyncio.run(agent.arun("..."))            # await agent.arun(...)
# agent.ask(...) / agent.aask(...) are aliases that read naturally with RAG

2. Providers

One API, many models. Switch with a string:

Agent(provider="openai",    model="gpt-4.1")
Agent(provider="anthropic", model="claude-opus-4-8")
Agent(provider="gemini",    model="gemini-2.0-flash")
Agent(provider="groq",      model="llama-3.3-70b-versatile")
Agent(provider="echo")      # offline, no key — for tests & demos

OpenAI-compatible endpoints (Azure OpenAI, OpenRouter, Together, vLLM, …) work via base_url. You can also register a custom provider:

from neuroagent.providers import register_provider, LLMProvider
# class MyProvider(LLMProvider): ...   (implement complete/stream/embed)
register_provider("myprovider", MyProvider)
Agent(provider="myprovider")

3. Tools

Decorate any function with @tool. The agent decides when to call it, runs it, and continues reasoning with the result. Sync and async tools are supported.

from neuroagent import Agent, tool
import httpx

@tool
def get_weather(city: str) -> str:
    """Return the current weather for a city."""
    return f"It's 22°C and sunny in {city}."

@tool(read_only=True)
async def search_docs(query: str) -> str:
    """Search internal documentation."""
    ...

agent = Agent(provider="openai", tools=[get_weather, search_docs])
print(agent.run("What should I wear in Paris today?").content)

# Add tools after construction:
agent.add_tool(get_weather)

4. Memory

Give an agent multi-turn recall. Pass a backend name or an instance; the agent loads prior turns into context and persists each turn automatically.

from neuroagent import Agent

agent = Agent(provider="openai", memory="sqlite")   # persists across restarts
agent.run("Hi, I'm Amit.")
agent.run("What's my name?")        # → "Your name is Amit."

Backend	memory=	Notes
In-process	"session"	Fast, ephemeral
SQLite	"sqlite"	Persistent, stdlib, multi-conversation
Redis	"redis"	Distributed (needs [redis] extra)
Vector	VectorMemory(embedder=...)	Semantic search over past messages

from neuroagent.memory import SQLiteMemory, VectorMemory
from neuroagent.providers.openai import OpenAIProvider

# Explicit instances for full control:
agent = Agent(provider="openai", memory=SQLiteMemory("chat.db", conversation_id="amit"))

# Semantic recall: find the most relevant prior messages, not just the latest.
mem = VectorMemory(embedder=OpenAIProvider())
agent = Agent(provider="openai", memory=mem)
relevant = await mem.search("what did we decide about pricing?", k=3)

5. RAG (chat with your documents)

Load documents and the agent answers grounded in them — load → chunk → embed → store → retrieve → inject, all built in.

from neuroagent import Agent

agent = Agent(provider="openai", rag=True)

agent.load_document("handbook.pdf")     # PDF/DOCX/TXT/CSV/MD/HTML/JSON (PDF/DOCX need [rag])
agent.load_text("Our refund window is 30 days.")

print(agent.ask("Summarize chapter 5 of the handbook.").content)
print(agent.ask("What is our refund window?").content)

# Better diversity in retrieved context with MMR re-ranking:
agent = Agent(provider="openai", rag=True, rag_k=6, rag_rerank=True)

Need a custom embedder or vector store? Build a KnowledgeBase and pass it in:

from neuroagent import Agent, KnowledgeBase
from neuroagent.providers.openai import OpenAIProvider
from neuroagent.rag import InMemoryVectorStore

kb = KnowledgeBase(embedder=OpenAIProvider(), store=InMemoryVectorStore(), chunk_size=800)
await kb.aadd_path("manual.pdf")
agent = Agent(provider="anthropic", rag=kb)   # chat model and embedder can differ

6. Database agents (NL → SQL)

Ask questions of a SQL database in plain English. Read-only by default — writes need an explicit opt-in.

from neuroagent import DatabaseAgent

db = DatabaseAgent("sales.db", provider="openai")          # SQLite (stdlib)
# db = DatabaseAgent("postgresql://user:pw@host/db", provider="openai")   # needs [sql] extra

print(db.ask("Who are the top 5 customers by revenue this quarter?").content)

# Direct, schema-aware access (no LLM):
print(db.schema_text)                                       # what the model sees
result = db.query("SELECT name, total FROM customers ORDER BY total DESC LIMIT 5")
print(result.to_text())

Supported via SQLAlchemy: PostgreSQL, MySQL, SQL Server, Oracle, SQLite (drivers per database).

7. API agents

Turn any HTTP API into an agent. Read-only (GET/HEAD) by default.

from neuroagent import APIAgent

api = APIAgent(
    "https://api.example.com",
    provider="openai",
    description="Users and orders API.",
    auth_token="bearer-token",        # sent as Authorization: Bearer ...
)

print(api.ask("How many orders did user 42 place last month?").content)

# Direct call (no LLM):
print(api.request("GET", "/users/42"))

8. Workflows (DAG orchestration)

Compose agents (and any callables) into a dependency graph. Independent steps run in parallel; dependent steps run in order. Conditions skip steps, failures cascade, and steps can retry.

from neuroagent import Agent, Workflow

researcher = Agent(provider="openai", system_prompt="You research topics thoroughly.")
writer     = Agent(provider="openai", system_prompt="You write engaging prose.")

wf = Workflow("content-pipeline")
wf.add_agent_step("research", researcher, lambda ctx: f"Research: {ctx.get('topic')}")
wf.add_agent_step(
    "draft", writer,
    lambda ctx: f"Write a post based on:\n{ctx.result('research')}",
    depends_on=["research"],
)

result = wf.run(inputs={"topic": "AI agent frameworks"})
print(result.success)            # True
print(result["draft"])           # the writer's output

# Plain-function steps, conditions, retries, and checkpoint resume are all supported:
from neuroagent.workflow import RetryPolicy
wf.add_step("flaky", do_work, retry=RetryPolicy(max_attempts=3, delay=1.0))
wf.run(inputs={...}, checkpoint={"research": "...prior result..."})   # resume

9. Multi-agent teams

Specialized agents collaborating via a chosen pattern.

from neuroagent import Agent, Team, TeamMember

researcher = TeamMember("researcher", Agent(provider="openai"), "gathers facts")
writer     = TeamMember("writer",     Agent(provider="openai"), "writes prose")
reviewer   = TeamMember("reviewer",   Agent(provider="openai"), "checks quality")

team = Team([researcher, writer, reviewer], pattern="pipeline")
result = team.run("Write a launch post about our new feature.")
print(result.content)            # final answer
print(result.contributions)      # each member's output

pattern=	Behavior
"pipeline"	Sequential hand-off: researcher → writer → reviewer
"router"	A coordinator picks the single best member to handle the task
"debate"	All members answer in parallel, then a coordinator synthesizes
"hierarchical"	A lead delegates to workers, then synthesizes the result

Agents can also message each other directly via the in-process Channel (from neuroagent import Channel) — the foundation for agent-to-agent communication.

10. Streaming

Stream tokens as they arrive:

import asyncio
from neuroagent import Agent

async def main():
    agent = Agent(provider="openai")
    async for token in agent.astream("Tell me a short story about a robot."):
        print(token, end="", flush=True)

asyncio.run(main())

11. Observability & cost

Every run is traced and costed automatically — no setup.

from neuroagent import Agent

agent = Agent(provider="openai")
agent.run("Summarize the news.")
agent.run("Draft a reply.")

print(agent.stats())
# {'runs': 2, 'tokens': 1423, 'cost': 0.0042, 'cost_str': '$0.0042',
#  'by_model': {...}, 'by_provider': {...}, ...}

# Inspect the last run as a tree of spans:
from neuroagent.observability import ConsoleExporter
print(ConsoleExporter().export(agent.last_trace))

# Ship traces to any OpenTelemetry collector (OTLP/HTTP, no SDK required):
from neuroagent.observability import OTelExporter
OTelExporter().send(agent.last_trace, "http://localhost:4318/v1/traces")

Pricing tables are data-driven and overridable:

from neuroagent.cost import register_pricing
register_pricing("my-model", input_per_million=1.0, output_per_million=2.0)

12. Security (RBAC, audit, secrets, sandbox)

Enterprise guardrails as standalone primitives you can wire into your trust boundaries.

from neuroagent import RBAC, AuditLog, SecretStore, Sandbox

# Role-based access control (supports "*" and "ns:*" wildcards)
rbac = RBAC()
rbac.define_role("analyst", {"db:read", "api:get"})
rbac.assign("amit", "analyst")
rbac.require("amit", "db:read")          # raises SecurityError if not permitted

# Append-only audit log (optional JSONL file sink)
audit = AuditLog(sink_path="audit.jsonl")
audit.record("amit", "db:query", resource="customers", rows=42)

# Secret resolution + masking (encryption via the [security] extra)
secrets = SecretStore({"API_KEY": "sk-supersecret-1234"})
print(secrets.masked("API_KEY"))         # ***************1234

# Command policy guard for shell/code/SQL tools
sandbox = Sandbox(allow=["ls", "cat"])
sandbox.check("rm -rf /")                # raises SecurityError

13. Business integrations

Read business systems in natural language — built on the API-agent foundation, read-only by default.

from neuroagent import ShopifyIntegration, SalesforceIntegration, OdooIntegration, SAPIntegration

shopify = ShopifyIntegration("my-store", access_token="shpat_...")
print(shopify.list_products(limit=10))

salesforce = SalesforceIntegration("https://acme.my.salesforce.com", access_token="...")
print(salesforce.query("SELECT Id, Name FROM Account LIMIT 5"))

odoo = OdooIntegration("https://acme.odoo.com", db="acme", username="admin", password="...")
print(odoo.search_read("res.partner", fields=["name", "email"]))

# Or hand any integration an LLM and ask in plain English:
agent = shopify.agent(provider="openai")
print(agent.ask("Which product has the most inventory?").content)

14. Deploy as an HTTP server

Turn any agent into an HTTP service with no FastAPI boilerplate ([server] extra).

from neuroagent import Agent

Agent(provider="openai").serve(port=8000)
#   POST /chat, /ask, /run   →  {"content": ..., "trace_id": ..., "tokens": ...}
#   POST /stream             →  server-sent events (token streaming)
#   GET  /health, /stats

curl -X POST localhost:8000/run -H "content-type: application/json" \
     -d '{"input": "Hello!"}'

Need the app object for your own server/middleware? Use agent.create_app() (returns a FastAPI app).

---

Putting it together

A research team whose members have tools, memory, and document knowledge — then served over HTTP:

from neuroagent import Agent, Team, TeamMember, tool

@tool
def web_search(query: str) -> str:
    """Search the web for up-to-date information."""
    ...

researcher = Agent(provider="openai", tools=[web_search], rag=True, memory="sqlite")
researcher.load_document("internal_strategy.pdf")

writer   = Agent(provider="anthropic", system_prompt="You write crisp executive summaries.")
reviewer = Agent(provider="openai",    system_prompt="You fact-check and tighten prose.")

team = Team(
    [
        TeamMember("researcher", researcher, "gathers internal + web context"),
        TeamMember("writer", writer, "drafts the summary"),
        TeamMember("reviewer", reviewer, "reviews for accuracy"),
    ],
    pattern="pipeline",
)

print(team.run("Brief the board on our AI strategy vs. competitors.").content)

More runnable examples (all offline) live in examples/: quickstart.py, memory_chat.py, rag_quickstart.py, database_agent.py, workflow_pipeline.py, team_collaboration.py, observability_cost.py, enterprise_security.py, serve_agent.py.

python examples/quickstart.py

---

Documentation

• docs/ARCHITECTURE.md — the finalized architecture: stack, package structure, core contracts (the ABCs), run lifecycle, and the Definition of Done.
• docs/spec/ — the phase-by-phase build record (what shipped, decisions, coverage).
• docs/IDEA.md — the product vision.
• CHANGELOG.md — release history.

Build the docs site locally: pip install "neuroagent-ai[all]" mkdocs-material && mkdocs serve.

---

Contributing & releasing

See CONTRIBUTING.md. Every change follows: branch per feature → small, focused commits → ruff + mypy --strict + pytest green → merge to main. Releases are automated — push a vX.Y.Z tag and CI builds, creates the GitHub release, and publishes to PyPI.

python -m venv .venv && source .venv/bin/activate    # Windows: .venv\Scripts\activate
pip install -e ".[dev]"
ruff check . && mypy src && pytest

---

License

Apache-2.0.