MangoAgent

MangoAgent is the central class that orchestrates the LLM ↔ tool loop. Every question goes through it.

Constructor

from mango import MangoAgent

agent = MangoAgent(
    llm_service=llm,         # required
    tool_registry=tools,     # required
    db=db,                   # required
    agent_memory=memory,     # optional
    schema=None,             # optional — pre-introspected schema
    introspect=False,        # whether setup() runs introspection
    max_iterations=8,        # safety cap on tool calls per question
    max_retries=2,           # retries on fixable tool errors
    memory_top_k=3,          # how many memory examples to inject
    training_top_k=3,        # how many verified training examples to inject
    max_turns=5,             # conversation turns to keep in history
    schema_top_k=3,          # collections detailed per question
    schema_always_all=4,     # detail every collection when the DB has ≤ N
    middlewares=None,        # access-control middlewares
)

Parameters

Methods

setup()

Initializes the agent: runs schema introspection (if introspect=True) and builds the system prompt. Call once after connecting the database, before the first ask().

agent.setup()

With introspect=False and no schema, setup() only builds the static system prompt (no database calls); the LLM then discovers collections through the tools.

ask(question, on_tool_call=None, *, user=None)

Ask a question and get a complete answer. user is an opaque object describing the caller (Mango never inspects it) that is handed to the access-control middlewares; ask() raises AccessDenied when a before_turn middleware blocks the turn.

response = await agent.ask("How many orders were placed last week?", user=current_user)
print(response.answer)

Optional on_tool_call callback receives (tool_name, tool_args, result_text) after each tool execution:

def log_tool(name, args, result):
    print(f"Tool: {name} | Args: {args}")

response = await agent.ask("...", on_tool_call=log_tool)

ask_stream(question, *, user=None)

Same as ask() but streams events as they happen via an async generator. Useful for real-time UIs. Takes the same keyword-only user and may raise AccessDenied as well.

async for event in agent.ask_stream("What are the top 10 products by revenue?"):
    if event["type"] == "tool_call":
        print(f"Calling {event['tool_name']}...")
    elif event["type"] == "answer":
        print(event["text"])

Event types:

new_session()

Returns a new agent with the same configuration but a fresh conversation history. Schema and system prompt are shared — no re-introspection.

session = agent.new_session()
response = await session.ask("...")

reset_conversation()

Clears the conversation history in place.

agent.reset_conversation()

Access control: use(), hook decorators, check_tool_access()

agent.use(middleware) registers a Middleware (runs in registration order). The decorators @agent.before_turn, @agent.filter_tools, @agent.before_tool, @agent.before_query, @agent.after_query, @agent.after_tool, @agent.before_answer and @agent.after_turn register single-hook middlewares from plain functions. agent.check_tool_access(user, tool_name, args=None) runs the before_tool hooks outside a turn (raising AccessDenied) so the host can gate non-agent actions such as the memory REST endpoints with the same policy. See Access Control.

Properties

Multi-turn conversations

Conversation history is preserved across ask() calls. Follow-up questions work naturally:

r1 = await agent.ask("How many orders were placed last week?")
r2 = await agent.ask("And how many of those were delivered?")
r3 = await agent.ask("Which customer placed the most?")

History is automatically pruned to max_turns complete turns to keep token usage stable.