from micro_agents import Aggregator, OpenAIBasedLLM, Tool, ToolManager


def current_weather(city: str) -> str:
    """Return the current weather for a city."""
    return f"Sunny in {city}"


class BotTools(ToolManager):
    async def is_allowed(self, tool, context: dict[str, object]) -> bool:
        return context.get("role") == "member"


def master_prompt(context: dict[str, object]) -> str:
    return f"You are a concise support bot. Reply in {context.get('language', 'en')}."


def user_prompt(context: dict[str, object]) -> str:
    return f"<user_message>\n{context['source_text']}\n</user_message>"


llm = OpenAIBasedLLM(api_url="https://api.openai.com/v1", api_key="...", model="gpt-5")
agent = Aggregator(
    llm,
    BotTools([Tool(current_weather)]),
    master_prompt=master_prompt,
    user_prompt=user_prompt,
)
session = agent.create_session(context={"user_id": 42, "role": "member", "language": "ru"})

result = await session.run("Какая погода в Варшаве?")
if result.succeeded:
    print(result.final_message.message())

Use async for event in session.stream(message) when the embedding needs tool-progress or message events. A session retains its conversation and a live application-owned context dict. ToolManager.is_allowed() is checked when tools are shown to the model and again immediately before execution.

An individual session can override the aggregator's defaults:

session = agent.create_session(llm=another_llm, thinking=True)

If llm is omitted, the aggregator's LLM is used. thinking=False disables reasoning, thinking=True requests medium reasoning, and the default None leaves the provider's behavior unchanged.

Attach a local image, document, or MP3/WAV file with Media.from_path(). Images and files may also use a URL or a provider file ID:

from micro_agents import Media

result = await session.run(
    "Что изображено и что написано в документе?",
    media=[Media.from_path("photo.png"), Media.from_path("notes.pdf")],
)

Do not interpolate secrets, API keys, or internal-only values into master_prompt: it receives the full live context dict.

user_prompt can synchronously or asynchronously normalize, wrap, or enrich every incoming user message before it is added to the conversation. It receives a copy of the session context with source_text set to the original text passed to run() or stream(); this reserved value overrides any source_text already present in the context without mutating the context itself. The returned string becomes the user message, while any media attachments stay attached to it.

Saving and restoring sessions

Save the application context and complete conversation, then create a new session from that state:

from micro_agents import SessionState

state = session.save()
await database.save(state.session_id, state.to_json())

loaded = SessionState.from_json(await database.load(state.session_id))
session = agent.create_session(state=loaded)

create_session() also accepts the result of state.to_dict() directly. Restoring preserves the session ID, optional name, messages, inline media, tool calls, tool responses, and provider reasoning items. The restored context and messages are fresh containers, so continuing the restored session does not mutate the saved state. Context values must be JSON-compatible when to_json() is used; pass a default converter for values such as Path, for example state.to_json(default=str).

Dynamic tools

Use DynamicToolManager when integrations can be connected or disconnected while the application is running:

from micro_agents import DynamicToolManager, Tool

tools = DynamicToolManager()
agent = Aggregator(llm, tools)

tools.add_tool(Tool(current_weather))
tools.remove_tool("current_weather")

Existing sessions read the current registry before every LLM round. Duplicate names are rejected unless add_tool(tool, replace=True) is used.

Context injection for tools

Mark a tool argument with Injected to receive the current session context. That argument is not included in the tool schema sent to the LLM, so the model cannot provide or override it.

from pathlib import Path
from typing import Annotated

from micro_agents import Injected, SessionContext


def read_note(path: str, context: Annotated[SessionContext, Injected]) -> str:
    return (Path(context["base_dir"].__str__()) / path).read_text()

The same aggregator may then create isolated sessions with different roots:

agent.create_session(context={"base_dir": Path("test/alice")})
agent.create_session(context={"base_dir": Path("test/bob")})

DeepSeek filesystem example

Install the optional tool package first:

uv add micro-agents-tools

The runnable module below reads DEEPSEEK_API_KEY, uses DeepSeek's OpenAI-compatible Responses API, and can only read or write files inside the current directory's test folder. Edit PROMPT in the file, then run it:

$env:DEEPSEEK_API_KEY = "..."
uv run --package micro-agents-tools -m micro_agents_tools.examples.deepseek_filesystem

Set DEEPSEEK_MODEL to override the default deepseek-v4-flash. Filesystem and opt-in Python tools are in the separate micro-agents-tools package. Filesystem tools deliberately have no delete operation, and reject absolute paths, .. escapes, and symlinks that resolve outside test.

PythonTools is opt-in: add *PythonTools().tools() to the manager only when the application trusts the code being generated. It runs a subprocess in base_dir with a timeout and bounded output, but it is not a security sandbox; use a container or VM for untrusted code.

FilesystemTools also exposes attach_files(paths). The model can select images or documents under the session's base_dir; they are returned as real tool-output attachments and become available to the model on the next tool-loop round. Paths remain confined to base_dir, and the combined attachment size is limited to 50 MiB by default.

BraveSearchTools().tools() adds a web_search tool backed by Brave Search. Set BRAVE_SEARCH_API_KEY in the application environment; the key never becomes a model argument or a session-context value.