Agent

#Agent

Agent is the primary single-agent scope. It owns credentials/client wiring, tools, skills/resources, runtime defaults, and invocation methods.

#Construction

from maivn import Agent, MemoryConfig, SystemToolsConfig

agent = Agent(
    name='researcher',
    description='Researches a question and cites evidence.',
    system_prompt='Prefer primary sources.',
    api_key='your-api-key',
    model='auto',
    memory_config=MemoryConfig(enabled=True),
    system_tools_config=SystemToolsConfig(
        allowed_tools=['web_search', 'repl'],
    ),
)

name is required. Credentials can come from api_key, the terminal environment, an explicitly selected env_file or api_key_file, or a preconfigured Client. See Authentication for precedence. Other constructor controls include timeout, max_results, private_data, system/orchestration config, final-output/nested-synthesis flags, hooks, tools, skills, resources, tags, skill sets, attached skill IDs, and MCP servers. Orchestration config is a retained compatibility model. max_results is also retained but does not limit v2 tool selection. Invocation metadata and per-call allow_private_in_system_tools are accepted but ignored. Use system_tools_config for per-call private-data policy.

agent_id defaults to name, and version defaults to 1. Trigger builders inherit that deployed identity; .target(...) remains an explicit override.

allow_private_in_system_tools defaults to False. Enabling it permits raw private values to reach system tools and should be an explicit risk decision. Privacy controls reduce exposure but are not certifications or guarantees; see Private Data.

#Tool registration

@agent.toolify(description='Look up an order')
def lookup_order(order_id: str) -> dict[str, str]:
    return {'order_id': order_id, 'state': 'ready'}

Use add_tool(...) for an existing function/model and add_toolset(...) for a connector instance. list_tools() returns registered ToolMetadata; compile_tools() returns the current compiled metadata and schemas; validate_tool_configuration() checks final-output routing.

toolify and add_tool accept name, description, permissions, destructive/always-execute/final-tool flags, metadata, tags, and before/after hooks. add_toolset filters by method names or tags.

#Invocation

The sync invocation signature is:

def invoke(
    messages: SdkMessagesInput,
    *,
    force_final_tool: bool | None = None,
    targeted_tools: list[str] | None = None,
    structured_output: type[BaseModel] | None = None,
    thread_id: str | None = None,
    verbose: bool = False,
    reasoning: Literal['minimal', 'low', 'medium', 'high'] | None = None,
    metadata: Mapping[str, object] | None = None,
    memory_config: MemoryConfig | Mapping[str, object] | None = None,
    allow_private_in_system_tools: bool | None = None,
    model: RuntimeModelChoice | None = None,
    system_tools_config: SystemToolsConfig | Mapping[str, object] | None = None,
    orchestration_config: SessionOrchestrationConfig | Mapping[str, object] | None = None,
    stream_response: bool = False,
    options: RunOptions | None = None,
    origin: str | None = None,
) -> InvokeResponse:
    ...

ainvoke(...) has the same keyword surface and returns an awaitable InvokeResponse. stream(...) and astream(...) have the same controls and yield StreamEvent values.

from maivn.messages import HumanMessage

response = agent.invoke(
    [HumanMessage(content='Find the current order state.')],
    targeted_tools=['lookup_order'],
    thread_id='thread-42',
)
print(response.response)

InvokeResponse contains final_message, response, optional result, session/thread/root-event identifiers, event positions, usage/token metadata, stop reason, and tool-call summary. It has no status or error field.

stream_response=True asks the SDK to consume the event stream internally while still returning one InvokeResponse; use stream() when the caller needs each event.

#Thread handles

thread(thread_id=None, *, options=None) binds a conversation without sending a request. Use send / asend for completed responses and stream / astream for events, with the same tools, hooks, message lists, and invocation arguments as this agent:

conversation = agent.thread()
response = conversation.send('Remember that the order is 42.')
response = conversation.send('Look up that order.', targeted_tools=['lookup_order'])

Use state / astate for state and events / aevents for current-turn replay. The handle fixes the thread ID. See Sessions and Streaming for cursor lifetime, explicit dispatch methods, and stream cleanup.

#Structured output

from pydantic import BaseModel


class Summary(BaseModel):
    answer: str


response = agent.structured_output(Summary).invoke('Summarize the order.')
summary = response.result

The structured_output(Model) builder preserves invoke/ainvoke/stream/astream. Direct structured_output=Model remains available on invocation methods. force_final_tool targets a registered final_tool=True tool and is a different mechanism.

#Final tool selection

When a run produces several final-tool calls, the first successful call supplies the final value. Calls published together retain their authored order regardless of parallel completion timing. Later calls cannot replace the selected value; fatal run errors still terminate the run as errors.

The final event's message metadata exposes three flat fields:

• final_tool_policy: "first_successful".
• final_tool_call_count: the cumulative number of final-tool calls counted by the run.
• final_tool_selected_call_id: the call ID that supplied the selected final value.

If bounded final-value generation and repair exhaust their attempts, the run terminates with value_fill_exhausted. invoke raises MaivnSDKError carrying that error_code, plus the session and root event IDs for diagnostics. Native final-tool prompting is also bounded: a model that continues returning prose after three reminders terminates with final_tool_contract_unfulfilled.

#Follow-up questions

from maivn import FollowupQuestion


def answer(question: FollowupQuestion, schema: dict[str, object]) -> object:
    print(question.question)
    return 'US East'


response = agent.allow_followup_questions(
    input_handler=answer,
    max_questions=3,
    max_questions_per_thread=10,
    response_schema={'type': 'string'},
    timeout=60,
).invoke('Plan the deployment.', thread_id='thread-42')

The builder handles runtime-authored agent_followup checkpoints inline, submits the typed answer, and continues the same thread. The default handler is default_terminal_followup. See Interrupts and follow-up questions.

#System tools and model routing

SystemToolsConfig.allowed_tools is the invocation allowlist. web_search retrieves current public information; repl runs sandboxed Python for calculation and analysis. Availability remains subject to workspace policy and plan limits.

model accepts a tier/directive or ModelConfig. A scoped ModelConfig can select response, thinking, compose_argument, and repl through ModelChoice, and planning through PlanningChoice. Omitted parts use platform routing. See model configuration.

from maivn import ModelChoice, ModelConfig

analyst = Agent(name='analyst', api_key='your-api-key')
response = analyst.invoke(
    'Analyze the dataset.',
    model=ModelConfig(
        response=ModelChoice(tier='balanced'),
        repl=ModelChoice(tier='fast'),
    ),
)

#Events and terminal output

events(include=..., exclude=..., on_event=..., auto_verbose=True) returns an invocation wrapper that reports selected compatibility events. verbose=True on invocation methods enables local terminal reporting.

A plain install uses Tier 0 text output. maivn[rich] enables Tier 1 Rich output on a capable terminal. MAIVN_TERMINAL_TIER=plain|0 forces plain; rich|1 requests Rich but degrades to plain if the optional dependency is unavailable. CI, NO_COLOR, TERM=dumb, and non-TTY output also select plain unless a supported explicit Rich override applies.

#Batch, scheduling, and cleanup

batch(inputs, max_concurrency=..., **invoke_kwargs) and abatch(...) preserve input order and return list[InvokeResponse].

cron(expression, tz=..., mode='remote', ...), interval(...), and at(datetime, ...) create invocation builders; see Scheduling. Agent.close() is currently a compatibility no-op. Close an owned Client and registered MCPServer instances explicitly; see MCP lifecycle.

See Tools, Sessions and Streaming, and Shared Data Models.