marona 0.10.3

Provider-neutral AI agent runtime and MCP/Skill gateway for Marona Hub connections, managed agents, and bring-your-own-agent integrations.

Marona Python SDK

Provider-neutral AI agent runtime and MCP/Skill gateway for Marona Hub connections, managed agents, and bring-your-own-agent integrations.

pip install marona

1. Marona Hub

Connect Apps and governed Skills as one neutral MCP tool collection.

from marona import Marona

marona = Marona(api_key="YOUR_MARONA_API_KEY")

# Connect every App and Skill available to this developer key.
connection = marona.hub.connect()

Select a smaller capability set when needed:

connection = marona.hub.connect(
    apps=["group-fund"],
    skills=["create-group-fund"],
)

Use the connection directly:

tools = connection.list_tools()
result = connection.call_tool(
    "skill__create_group_fund",
    {"request": "Create a family savings group fund"},
)

MCPConnection is not tied to OpenAI, LangGraph, CrewAI, or another model vendor. It exposes:

connection.list_tools()
connection.call_tool(name, arguments)
connection.server_url
connection.server_urls
connection.warnings
connection.session

An unresolved App or Skill name does not discard valid tools. Check connection.warnings for its code, selector_type, slug, and corrective message. Authentication, permission, and configured-server failures remain blocking errors.

Marona Hub owns discovery, identity, permissions, App and Skill resolution, approvals, governed execution, and online, offline, or hybrid availability. With no selectors, online and hybrid connections include every capability available to the developer key; offline connections include every capability installed on the device.

Connected Apps execute through Marona governance. When an App such as Google Contacts or Calendar needs sign-in, the tool result includes structuredContent.authorization_url and a link artifact. Pass user_id when calling a tool directly so the resulting service connection belongs to the correct end user:

result = connection.call_tool(
    "contacts__search_contacts",
    {"query": "John"},
    user_id="customer_482",
    conversation_id="chat_91a7",
)

Use the asynchronous methods inside an event loop:

connection = await marona.hub.connect_async(
    apps=["group-fund"],
    skills=["create-group-fund"],
)
tools = await connection.list_tools_async()
result = await connection.call_tool_async(
    "skill__create_group_fund",
    {"request": "Create a family savings group fund"},
)

2. Marona Agent

Use Marona's Agent and Runner when you want one simple managed agent API.

from marona import Agent, Marona, Runner

marona = Marona(api_key="YOUR_MARONA_API_KEY")

tools = marona.hub.connect(
    apps=["sda-books"],
)

agent = Agent(
    name="Customer Assistant",
    model="marona/gpt-5.6",
    instructions="Help the customer.",
    tools=tools,
    tool_choice="auto",
)

result = await Runner.run(
    agent,
    "Download Steps to Christ",
    user_id="customer_482",
    session_id="chat_91a7",
)

print(result.output)

user_id is optional. session_id is also optional and defaults to default. Marona derives the developer scope from the authenticated API key and keeps conversation history isolated by developer, user, and session. In synchronous applications, use Runner.run_sync(...).

tool_choice defaults to "auto". Use "required" when the first model turn must select a connected capability, or "none" to disable tools for the Agent. After discovery Marona requires one exact capability call, then returns to automatic selection so multi-step requests can continue.

Automatic Model And Provider Routing

Use marona/auto for capability-aware model and provider routing. Tools, voice behavior, input restrictions, output restrictions, and routing are all optional:

from marona import Agent, Marona, Runner

marona = Marona(api_key="YOUR_MARONA_API_KEY")
tools = marona.hub.connect()

agent = Agent(
    name="Customer Assistant",
    model="marona/auto",
    instructions=(
        "Help customers clearly and concisely. "
        "Use available tools only when necessary."
    ),
    tools=tools,
    voice={
        "output_voice": "ash",
        "language": "en",
        "turn_detection": "semantic_vad",
        "interruptible": True,
        "input_format": "pcm16",
        "output_format": "pcm16",
    },
    modalities=["text", "image", "audio"],
    output_modalities=["text", "audio"],
    routing={
        "strategy": "balanced",
        "max_cost_usd": 0.02,
        "target_latency_ms": 3000,
        "timeout_ms": 30000,
        "allowed_models": [
            "marona/gpt-5-mini",
            "marona/gemini-2.5-pro",
        ],
        "allowed_providers": [
            "openai",
            "google",
            "openrouter",
        ],
        "fallback": True,
    },
)

result = await Runner.run(
    agent,
    input=[
        {
            "role": "user",
            "content": [
                {
                    "type": "input_text",
                    "text": "Describe the problem shown in this image.",
                },
                {
                    "type": "input_image",
                    "image_url": "https://example.com/damaged-package.jpg",
                },
            ],
        }
    ],
    user_id="customer_482",
    session_id="chat_91a7",
)

print(result.output)

For ordinary conversation, no Hub connection is required:

agent = Agent(name="Customer Assistant")
result = await Runner.run(agent, "Hello")
print(result.output)

Text-only runs return a str. Mixed or media output returns ordered output_text, output_image, output_audio, output_video, and output_file parts. Permission, approval, and service-auth continuations raise MaronaPermissionRequired, MaronaApprovalRequired, or MaronaServiceConnectionRequired.

Streaming

async for event in Runner.stream(
    agent,
    input="Explain this document.",
    user_id="customer_482",
    session_id="chat_91a7",
):
    if event.type == "output_text.delta":
        print(event.delta, end="")

Realtime

from marona import RealtimeRunner

session = await RealtimeRunner.connect(
    agent,
    user_id="customer_482",
    session_id="voice_91a7",
)

async with session:
    await session.send_message("List my notes")
    async for event in session:
        if event.type == "transcript.delta":
            print(event.delta, end="")
        elif event.type == "response.completed":
            print(event.output)
            break

The persistent session also supports send_audio(...) and close(). Provider events, tool discovery, MCP/Skill execution, and function results are handled inside the session; applications receive only normalized Marona events.

Typed Input And Output

Use Pydantic models when an Agent needs a validated contract. Marona validates input before calling a model and returns the validated output model:

from pydantic import BaseModel
from marona import Agent, Runner

class CustomerRequest(BaseModel):
    question: str
    priority: str = "normal"

class CustomerResponse(BaseModel):
    answer: str
    needs_follow_up: bool

agent = Agent(
    name="Customer Assistant",
    model="marona/gpt-5-mini",
    input_type=CustomerRequest,
    output_type=CustomerResponse,
    instructions="Answer the customer's question.",
)

result = await Runner.run(
    agent,
    CustomerRequest(question="When will my order arrive?"),
)

print(result.output.answer)

Multi-Agent

Use Agent.as_tool() for bounded delegation. The parent remains active and combines the specialist result. Use handoff(...) when the destination should take ownership of the conversation:

from marona import Agent, Marona, Runner, handoff

marona = Marona(api_key="YOUR_MARONA_API_KEY")
tools = marona.hub.connect()

researcher = Agent(
    name="Research Specialist",
    description="Researches bounded customer questions.",
    model="marona/gpt-5-mini",
    tools=tools,
)

billing = Agent(
    name="Billing Specialist",
    description="Owns billing and payment conversations.",
    model="marona/gpt-5-mini",
    tools=tools,
)

manager = Agent(
    name="Customer Assistant",
    model="marona/gpt-5.6",
    tools=[
        researcher.as_tool(
            name="research",
            description="Research one bounded question and return the result.",
        ),
    ],
    handoffs=[
        handoff(
            billing,
            name="transfer_to_billing",
            description="Transfer billing and payment requests.",
        ),
    ],
)

result = await Runner.run(
    manager,
    "I was charged twice and need help.",
    user_id="customer_482",
    session_id="support_91a7",
)

print(result.output)
print(result.active_agent.name)
print(result.agent_runs)

The same manager works with Runner.stream(...) and RealtimeRunner.connect(...). Marona validates the complete graph before the first model call, rejects cycles, enforces execution limits, and keeps provider credentials out of agent and model context.

Realtime Multi-Agent

Define the complete Agent graph, then pass its manager to RealtimeRunner.connect(...). Delegation tools and handoffs remain available throughout the persistent realtime session:

from marona import Agent, Marona, RealtimeRunner, handoff

marona = Marona(api_key="YOUR_MARONA_API_KEY")
tools = marona.hub.connect()

researcher = Agent(
    name="Research Specialist",
    description="Researches bounded customer questions.",
    model="marona/gpt-5-mini",
    tools=tools,
)

billing = Agent(
    name="Billing Specialist",
    description="Owns billing and payment conversations.",
    model="marona/gpt-5-mini",
    tools=tools,
)

manager = Agent(
    name="Customer Assistant",
    model="marona/gpt-5.6",
    tools=[
        researcher.as_tool(
            name="research",
            description="Research one bounded question and return the result.",
        ),
    ],
    handoffs=[
        handoff(
            billing,
            name="transfer_to_billing",
            description="Transfer billing and payment requests.",
        ),
    ],
)

try:
    session = await RealtimeRunner.connect(
        manager,
        user_id="customer_482",
        session_id="support_voice_91a7",
    )

    async with session:
        await session.send_message("I was charged twice and need help.")

        async for event in session:
            if event.type == "agent.handoff.completed":
                print(event.data["target_agent"])
            elif event.type == "response.completed":
                print(event.output)
                break
finally:
    await marona.aclose()

Marona keeps provider events internal and emits one provider-neutral event stream. After a handoff, the destination Agent owns the following turns.

3. Marona Runtime

Use responses.create(...) when Marona should manage model reasoning, tool selection, permission and approval checks, execution, and the final response.

from marona import Marona

marona = Marona(api_key="YOUR_MARONA_API_KEY")

tools = marona.hub.connect(
    apps=["group-fund"],
    skills=["create-group-fund"],
)

response = marona.responses.create(
    model="marona/gpt-5.6",
    tools=tools,
    input="Create a family savings group fund",
    tool_choice="auto",
)

print(response.output)

model="marona/auto"
model="marona/gpt-5.6"
model="marona/gpt-5-mini"
model="marona/gpt-5-nano"
model="marona/claude-sonnet"
model="marona/gemini-2.5-pro"
model="marona/deepseek-chat"

Use marona/auto when Marona should select both the model and provider. Hard capability requirements are enforced first, a small internal classifier reasons about task type and complexity, and provider health, cost, and latency select where the chosen model runs:

response = marona.responses.create(
    model="marona/auto",
    routing={
        "strategy": "balanced",  # balanced | quality | cost | latency
        "max_cost_usd": 0.02,
        "target_latency_ms": 3000,
        "timeout_ms": 30000,
        "allowed_models": ["marona/gpt-5-mini", "marona/claude-sonnet"],
        "allowed_providers": ["openai", "anthropic", "openrouter"],
        "fallback": True,
    },
    input="Review this distributed system design.",
)

print(response.output)
print(response.model)
print(response.routing)

With a specific marona/... model, Marona keeps that exact model and may only change provider. Direct-provider routes are never rerouted.

The same request also supports direct-provider, private, and local models:

model="openai/gpt-5.6"                       # Marona key + OpenAI key
model="openrouter/anthropic/claude-sonnet"   # OpenRouter
model="anthropic/claude-sonnet"              # Anthropic directly
model="google/gemini"                        # Google directly
model="ollama/qwen3"                         # Ollama
model="litellm/local-qwen"                   # LiteLLM gateway
model="local/qwen"                           # Downloaded/in-process model

Direct provider routes resolve credentials from their standard environment variables. For OpenRouter, set OPENROUTER_API_KEY; Marona automatically uses https://openrouter.ai/api/v1 and preserves the remaining OpenRouter model slug:

response = marona.responses.create(
    model="openrouter/anthropic/claude-sonnet",
    input="Help me with this request",
)

print(response.output)

Register only custom providers or downloaded in-process models:

marona.models.register(
    name="office/company-assistant",
    endpoint="https://models.office.example/v1",
    model="company-assistant-v2",
    api_key="YOUR_PROVIDER_KEY",
)

marona.models.register(
    name="local/qwen",
    executor=qwen_executor,
    context_window=8192,
    max_output_tokens=512,
)

For asynchronous applications, call await marona.responses.create_async(...).

Images

from marona import Marona

marona = Marona(api_key="YOUR_MARONA_API_KEY")

response = marona.responses.create(
    model="openai/gpt-5.6",
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "Summarize this image."},
                {"type": "input_image", "image_url": "https://example.com/image.jpg"},
            ],
        }
    ],
)

print(response.output)

Files

from marona import Marona

marona = Marona(api_key="YOUR_MARONA_API_KEY")

response = marona.responses.create(
    model="openai/gpt-5.6",
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "Summarize this file."},
                {
                    "type": "input_file",
                    "filename": "report.pdf",
                    "file_data": "data:application/pdf;base64,...",
                    "detail": "high",
                },
            ],
        }
    ],
)

print(response.output)

4. Bring Your Own Agent

The external framework owns its Agent, reasoning, and orchestration. Marona supplies neutral MCP tools and retains authorization, approvals, and execution.

OpenAI Agents SDK Example

from marona import Marona
from agents import Agent, Runner

marona = Marona(api_key="YOUR_MARONA_API_KEY")
connection = marona.hub.connect(
    apps=["group-fund"],
    skills=["create-group-fund"],
)

# Adapt only at the framework boundary. Marona itself remains vendor-neutral.
framework_tools = your_openai_agents_mcp_adapter(connection)

agent = Agent(
    name="Group Fund Assistant",
    model="marona/gpt-5.6",
    instructions="Help users create and manage group funds.",
    tools=framework_tools,
)

result = Runner.run_sync(agent, "Create a family savings group fund")
print(result.final_output)

your_openai_agents_mcp_adapter(...) represents the OpenAI-specific adapter at the framework boundary; it is not part of Marona's vendor-neutral core API.

An MCP-compatible framework can map its standard tool-list and tool-call hooks directly to connection.list_tools() and connection.call_tool(...). Marona does not claim that one Python tool object automatically satisfies every agent framework's proprietary interface.

8. Publish A Skill

Every workflow entry uses step(); type selects reasoning, approval, or App execution. Set visibility="public" for Hub discovery or "private" for only the owning developer workspace. New Skills default to private.

from marona import Marona
from marona.skills import skill, step


@skill(
    name="create-group-fund",
    description="Create a group fund after explicit user approval.",
    visibility="public",
    governs=["group-fund.create_group"],
)
def create_group_fund():
    request = step(
        id="understand-request",
        type="reasoning",
        instruction="Extract the group name and currency.",
        inputs={"message": "{{ context.user_message }}"},
        outputs={"name": "string", "currency": "string"},
    )
    permission = step(
        id="confirm-create",
        type="approval",
        message=f"Create '{request.name}' in {request.currency}?",
        outputs={"approved": "boolean"},
    )
    return step(
        id="create-group",
        type="app",
        app="group-fund",
        capability="group-fund.create_group",
        instruction="Create the approved group.",
        condition=permission.approved,
        inputs={"name": request.name, "currency": request.currency},
        outputs={"group_id": "string", "name": "string"},
    )


marona = Marona(api_key="YOUR_MARONA_API_KEY")
marona.skills.publish(create_group_fund, version="1.0.0")

Execution Modes

Set mode when creating Marona:

marona = Marona(
    api_key="YOUR_MARONA_API_KEY",
    mode="hybrid",
)

• online: network models and online MCP targets are allowed.
• hybrid: local/private execution may fall back to online execution.
• offline: only installed local Apps, Skills, data, and local models run.

Changing model never changes App, Skill, permission, approval, or MCP rules.