Models & tools

Register a Python function

ToolRegistry.tool generates an argument schema from supported Python annotations. The function remains callable. Declare effect explicitly; annotation inference grants no authorization.

from ralf_agent import ToolRegistry

tools = ToolRegistry()

@tools.tool(effect="read_only", description="Read a ticket")
def lookup(ticket: str):
    return {"ticket": ticket, "status": "open"}

Complex types or constraints require an explicit schema or custom adapter. Defaults are validated and captured in registration identity. Register a new definition and use new-task configuration when that definition changes.

What happens in a tool round trip

1. The agent declares allowed tool names; the provider receives descriptions and model context.
2. The model proposes a tool name and JSON arguments. The framework checks schemas, allowed scope, allowances and host authorization.
3. If approval is required, it records the original action and digest and pauses. After approval, the executor calls that tool.
4. The result becomes a receipt and enters subsequent model context. The model continues or returns final output.

A Python function remains directly callable; this execution evidence belongs to the framework-dispatched path. Registration neither exposes it to every agent nor places business code in an isolated environment.

Connect a model and run

Set OPENAI_API_KEY and RALF_MODEL in the process environment. Choose a model actually available to your account. Save this complete example as ticket_agent.py in a new working directory. It uses a real provider and incurs provider usage; use the offline application test first if you want a key-free tool loop.

import asyncio
import os
from ralf_agent import (
    AgentSpec, LocalEngine, OpenAIChatProvider,
    SQLiteStore, ToolRegistry,
)

tools = ToolRegistry()

@tools.tool(effect="read_only", description="Read a ticket")
def lookup(ticket: str):
    return {"ticket": ticket, "status": "open"}

spec = AgentSpec(
    "assistant", "Read the ticket and summarize its status.",
    ("lookup",),
)

async def main():
    provider = OpenAIChatProvider(model=os.environ["RALF_MODEL"])
    with LocalEngine(
        SQLiteStore(".ralf/agent.sqlite"),
        providers={"default": provider}, tools=tools,
    ) as engine:
        run = engine.submit(spec, "Check INC-001",
                            command_id="request-001", budget=16)
        print(await engine.drive(run.run_id))

asyncio.run(main())

Run python ticket_agent.py. A successful response prints a snapshot with state='succeeded' and a ticket summary. Model behavior can differ; inspect failures using approval and recovery and check the actual tool receipts rather than assuming the model called lookup. Repeating this exact command retrieves the original run; new input or configuration needs a new command ID.

Here spec is an AgentSpec declaration. The engine submits and drives it. The complete Session example instead creates an Agent entry that provides open_session; these are different objects.

The provider proposes responses and calls. The engine controls authorization and execution. Each compatible endpoint needs its own protocol and tool-call verification. See the model adapter guide (Chinese) for more adapters.

Constrain final output

A tool argument schema constrains inputs. The agent’s output_schema constrains its final result. Replace the preceding spec definition with:

spec = AgentSpec(
    "assistant", "Read the ticket. Return ticket and status as JSON.",
    ("lookup",),
    output_schema={
        "type": "object",
        "properties": {
            "ticket": {"type": "string"},
            "status": {"type": "string"},
        },
        "required": ["ticket", "status"],
        "additionalProperties": False,
    },
)

Also change the submission to command_id="request-structured-001", then run python ticket_agent.py again. The changed definition is a new request: reusing request-001 with it raises IdempotencyConflict. Keep the database and original definition for recovering the old run. An already recorded response is not regenerated by rerunning a command.

Local validation and a provider’s native structured output are separate layers. Configure native output explicitly where the adapter and model support it. Valid JSON does not prove the ticket was handled. Completion rules such as ToolReceiptVerifier check the current node’s tool receipts. See completion verification (Chinese).

Declare write boundaries

A write tool can declare requires_approval=True. To receive host context, explicitly set context_parameter. The executor injects it; the model cannot supply it.

@tools.tool(
    effect="idempotent_write",
    requires_approval=True,
    context_parameter="context",
    version="1",
)
def apply(value: int, *, context):
    return business_service.apply(
        value, idempotency_key=context.effect_key,
    )

Your application supplies business_service. The downstream system must actually honor its idempotency or query contract. Declaring an effect category does not create a cross-system exactly-once guarantee.

The model cannot supply context.effect_key. Timeouts, cancellation and exceptions do not establish that a business operation never ran. Retain the original action and recover or reconcile according to its effect category.

Preserve identity and lifecycle

Keep credentials in the environment or trusted host. Changed identities of tools, providers, context or memory belong to new tasks; existing runs still need their original configuration. Python extensions are trusted host code. Configure isolation, permissions and shutdown explicitly in your deployment.

Use integrations and extensions for existing MCP, HTTP or CLI tools. Read approval and recovery to handle writes across timeouts and process exits.

Check signatures and return values in API & CLI reference, verify an endpoint in testing and evaluation, or continue one application through models, sessions and knowledge in the application tutorial.