Reference
Best Practices
Production-grade patterns for robust agent behavior.
#Best Practices
A scannable set of recommended patterns for building robust mAIvn agents in production, grouped by concern: agent design, tools, dependencies, orchestration, security, errors, performance, testing, logging, and swarms.
#Agent Design
#Clear Naming
Use descriptive names that reflect purpose:
# Goodagent = Agent( name='customer_support_agent', description='Handles customer inquiries and support tickets', api_key='...',) # Avoidagent = Agent(name='agent1', api_key='...')#Effective System Prompts
Write system prompts that guide behavior:
agent = Agent( name='data_analyst', system_prompt='''You are a data analyst assistant. Your capabilities: - Analyze datasets using the analyze_data tool - Generate reports using the create_report tool - Answer questions about data trends Guidelines: - Always validate data before analysis - Explain your methodology clearly - Provide confidence levels for predictions - Use the final_report tool for all final outputs''', api_key='...',)#Focused Agents
Each agent should have a clear, focused responsibility:
# Good: focused agentsresearcher = Agent(name='researcher', description='Finds information', api_key='...')analyst = Agent(name='analyst', description='Analyzes data', api_key='...')writer = Agent(name='writer', description='Writes content', api_key='...') # Avoid: unfocused agentgeneral = Agent(name='everything', description='Does all tasks', api_key='...')#Tool Design
#Choose the Registration Style That Fits the Module
Use @agent.toolify(...) when the tool naturally belongs next to the agent definition:
@agent.toolify(description='Fetch customer orders')def get_orders(customer_id: str) -> dict: ...Use Agent(..., tools=[...]) when tools are already defined and you want the agent
configuration to show its complete tool surface:
def get_orders(customer_id: str) -> dict: """Fetch customer orders.""" ... agent = Agent( name='orders', api_key='...', tools=[get_orders],)Use agent.add_tool(...) when tools are assembled conditionally or imported from another
module:
agent = Agent(name='orders', api_key='...')agent.add_tool(get_orders, description='Fetch customer orders')All three styles use the same underlying tool registry and dependency decorators.
#Clear Descriptions
Tool descriptions help the LLM decide when to use them:
# Good: specific and actionable@agent.toolify(description='Search product catalog by name, SKU, or category')def search_products(query: str) -> dict: ... # Avoid: vague@agent.toolify(description='Search')def search_products(query: str) -> dict: ...#Type Hints and Defaults
Use specific types and sensible defaults:
from typing import Literalfrom pydantic import Field @agent.toolify(description='Fetch customer orders')def get_orders( customer_id: str, status: Literal['pending', 'shipped', 'delivered'] | None = None, limit: int = Field(default=10, ge=1, le=100),) -> dict: ...#Focused Tools
One tool should do one thing well:
# Good: separate focused tools@agent.toolify()def fetch_user(user_id: str) -> dict: ... @agent.toolify()def update_user(user_id: str, data: dict) -> dict: ... @agent.toolify()def delete_user(user_id: str) -> dict: ... # Avoid: multi-purpose tool@agent.toolify()def manage_user(action: str, user_id: str, data: dict = None) -> dict: ...#Agent-Ready Toolsets
A toolset should be useful when a developer registers it directly and asks the
agent for a normal workflow:
agent.add_toolset(MyToolSet(...))If a toolset only works after custom DAG construction, long prompt
engineering, or demo-specific wrappers, treat that as an interface problem.
Move the needed guidance into the tool names, descriptions, parameter schemas,
defaults, permission markers, and return values.
For provider-style toolsets:
- Use
@toolsetfor related operations that share state, credentials, or an
HTTP client. - Put accurate
permissions=PermissionSet(...)on every@toolifymethod so
callers can useinclude_tags=['read']andexclude_tags=['destructive']. - Mark irreversible tools with
destructive=True. - Return compact, human-readable summaries by default from broad list/search
tools. - Hide raw provider IDs by default unless another tool needs them; prefer an
opt-ininclude_ids=Falseargument. - Cap broad searches and expensive metadata expansion with small useful
defaults. - Let write tools accept natural read-tool outputs when the mapping is safe and
obvious. - Keep provider toolsets generic and use
ToolOverridefor app-specific names,
descriptions, default args, dependencies, or final-tool behavior.
#Human-Readable Tool Results
Agent final answers get worse when tool results are mostly internal handles.
Prefer result shapes like this:
{ 'tickets': [ { 'ticket_ref': 'ticket_1', 'title': 'Cannot sign in', 'requester': 'Ada Lovelace', 'status': 'open', 'updated_at': '2026-05-16T13:30:00Z', } ], 'next_page_token': '...',}Use raw IDs for follow-up tool calls, not as the main content in final
responses. If a downstream update tool needs the provider ID, expose it only
when requested:
def search_tickets(query: str, *, limit: int = 10, include_ids: bool = False) -> dict: ...#Error Returns
Return structured errors instead of raising exceptions:
@agent.toolify()def fetch_data(source: str) -> dict: try: data = external_api.fetch(source) return {'status': 'success', 'data': data} except NotFoundError: return {'status': 'error', 'error': f'Source not found: {source}'} except PermissionError: return {'status': 'error', 'error': 'Permission denied'}#Dependency Management
#Parallel Where Possible
Design for parallel execution:
# Good: independent tools run in parallel@agent.toolify()def fetch_users() -> dict: ... @agent.toolify()def fetch_products() -> dict: ... @agent.toolify()@depends_on_tool(fetch_users, 'users')@depends_on_tool(fetch_products, 'products') # Both run in paralleldef generate_report(users: dict, products: dict) -> dict: ...#Orchestration Policy
#Choose Single-Shot or Supervised Execution Explicitly
Use SessionOrchestrationConfig on both Agent and Swarm runs when the default
one-batch behavior is not enough:
from maivn import SessionOrchestrationConfig repair_policy = SessionOrchestrationConfig( mode='supervisor_loop', final_output_mode='supervised', allow_followup_actions=True, stop_strategy='objective_satisfied', max_cycles=5,)Good defaults by workflow:
- Simple Q&A / read-only reports: keep
single_shot_dagand terminal final output. - Repair, cleanup, and validation loops: use
supervisor_loop, supervised final output, and an objective-based stop strategy. - User-supplied exact DAGs: use
strict_user_dagand disable follow-up actions unless recovery is explicitly desired. - Developer-supplied first step with autonomous recovery: use
hybrid.
#Treat Final Output as a Role, Not Always a Stop Condition
For swarms, use_as_final_output=True identifies the member that can produce the
user-facing response. It should be terminal only for reporting workflows. For repair or
verification-heavy workflows, configure final_output_mode='supervised' so the
orchestrator can inspect the report and schedule more actions if the objective is not
satisfied.
#Security
#Private Data for Secrets
Never hardcode secrets:
# Good: use private dataagent.private_data = {'api_key': os.environ['API_KEY']} @agent.toolify()@depends_on_private_data(data_key='api_key', arg_name='key')def call_api(key: str) -> dict: ... # Avoid: hardcoded secrets@agent.toolify()def call_api() -> dict: api_key = 'sk-xxx-hardcoded' # Never do this! ...#Minimal Private Data
Only include what's needed:
# Good: only what's neededagent.private_data = { 'db_connection_string': '...', # Used by database tool} # Avoid: everything "just in case"agent.private_data = { 'db_connection_string': '...', 'unused_secret_1': '...', 'unused_secret_2': '...',}#PII Whitelist: Use Sparingly, Justify Always
PIIWhitelist lets a RedactedMessage opt out of redacting specific
spans (public URLs, your own published support address, etc.). Treat it
as a narrow exception, not a default:
from maivn import PIIWhitelist, PIIWhitelistEntry, RedactedMessage # Good: narrow value entry, real justification, PHI-safe.RedactedMessage( content='Reach us at support@maivn.io.', pii_whitelist=PIIWhitelist( entries=[ PIIWhitelistEntry( value='support@maivn.io', justification='Public support address listed on docs site.', ), ], phi_mode=False, ),) # Avoid: broad entity_type whitelist with weak justification.PIIWhitelist( entries=[ PIIWhitelistEntry(entity_type='email', justification='easier'), ],)#Always Set `phi_mode=True` in PHI Deployments
If your deployment ever handles Protected Health Information, setphi_mode=True on every PIIWhitelist. The validator will refuse any
entity-type whitelist for HIPAA Safe Harbor categories at construction
time:
from maivn import HIPAA_SAFE_HARBOR_CATEGORIES, PIIWhitelist, PIIWhitelistEntry # Validate at boot rather than at request time.def build_whitelist(entries: list[PIIWhitelistEntry]) -> PIIWhitelist: for entry in entries: if entry.entity_type and entry.entity_type in HIPAA_SAFE_HARBOR_CATEGORIES: raise ValueError( f"Refusing PHI-unsafe whitelist entry: {entry.entity_type}" ) return PIIWhitelist(entries=entries, phi_mode=True)#Always Provide a Real `justification`
The justification is required (≥8 chars) and is recorded as compliance
evidence for every suppressed span. Auditors read it. Write a real
reason — "approved by legal", "public address per RFC 1234", "B2B
research session, no PHI in scope".
#Match EventBridge Audience to the Frontend
Treat browser event streams as a trust boundary:
from maivn.events import EventBridge # End-user browser UIpublic_bridge = EventBridge('session-1', audience='frontend_safe') # Internal developer/admin UIinternal_bridge = EventBridge('session-1', audience='internal')Use frontend_safe when the event stream reaches end users. Use internal only for trusted tooling such as mAIvn Studio or your own internal operator consoles.
For the easiest possible end-to-end setup, use the FastAPI adapter — one line wires GET /maivn/events/{session_id} and your frontend (in any language) can consume it via EventSource or any SSE client:
from fastapi import FastAPIfrom maivn.events.fastapi import mount_events app = FastAPI()mount_events(app, factory=lambda sid: EventBridge(sid, audience='frontend_safe'))See the frontend events guide for client examples in JavaScript, TypeScript, Swift, Kotlin, Go, Python, Rust, and more.
#Harden Third-Party stdio MCP Servers
If you launch third-party MCP servers over stdio, prefer explicit environment passing instead of inheriting your full backend environment:
mcp_server = MCPServer( name='external_tools', transport='stdio', command='python', args=['-m', 'my_mcp_server'], inherit_env_allowlist=['OPENAI_API_KEY'], env={'SERVICE_TOKEN': 'explicit-token'}, stdio_response_timeout_seconds=30,)The SDK defaults to a minimal subprocess environment. Set inherit_env=True only when you trust the MCP server and it genuinely needs the full parent shell environment.
#Error Handling
#Graceful Degradation
Handle errors without crashing:
@agent.toolify()def fetch_with_fallback(primary_source: str) -> dict: try: return {'data': fetch_from_primary(primary_source)} except PrimarySourceError: try: return {'data': fetch_from_backup(), 'source': 'backup'} except BackupError: return {'error': 'All sources unavailable'}#Informative Errors
Return errors that help debugging:
@agent.toolify()def process_data(data: dict) -> dict: if 'required_field' not in data: return { 'error': 'Missing required_field', 'received_fields': list(data.keys()), } ...#Performance
#Timeout Configuration
Set appropriate timeouts:
client = Client( api_key='...', tool_execution_timeout=900, # 15 min per tool dependency_wait_timeout=300, # 5 min for deps total_execution_timeout=7200, # 2 hours total)#Limit Results
Use max_results for large tool catalogs (limits semantic search only; final/targeted tools and dependencies can add more):
agent = Agent( name='large_catalog_agent', max_results=15, # Limit semantic search results api_key='...',)#Resource Cleanup
Clean up resources when done:
agent = Agent(name='temp', api_key='...')try: response = agent.invoke([...])finally: agent.close()#Testing
Because a tool is just your function, the bulk of your coverage is ordinary Python —
call the function directly, no agent or network involved:
def test_fetch_data(): result = fetch_data('test-source') assert result['status'] == 'success' assert 'data' in resultFor inspecting the compiled request offline (compile_state()), mocking agent responses
and streams, and gating live tests behind a marker, see the dedicated
Testing & Local Development guide.
#Logging
#Enable in Development
from maivn import configure_logging logger = configure_logging('logs/dev.log')logger.setLevel('DEBUG')#Use Event Builder for Live Tracing
response = agent.events().invoke(messages)#Don't Log Secrets
# Goodlogger.info(f'Processing request for user {user_id}') # Avoidlogger.info(f'Using API key {api_key}') # Never log secrets!#Swarm Design
#Clear Responsibilities
Each agent in a swarm should have a distinct role:
swarm = Swarm( name='content_team', agents=[ Agent(name='researcher', description='Finds information', api_key='...'), Agent(name='analyst', description='Analyzes findings', api_key='...'), Agent(name='writer', description='Writes final content', api_key='...'), ],)#Single Final Output
Designate exactly one source of final output:
# Option 1: final output agentwriter = Agent(name='writer', api_key='...', use_as_final_output=True) # Option 2: swarm-level final tool@swarm.toolify(final_tool=True)class TeamReport(BaseModel): ...#Minimal Agent Count
Use the minimum number of agents needed:
# Good: 3 agents for clear separation of concernsresearcher -> analyst -> writer # Avoid: unnecessary agentsresearcher -> summarizer -> analyst -> formatter -> writer -> editor#See Also
- Troubleshooting - Solving common issues
- API Reference - Complete API documentation
- Guides - Step-by-step tutorials