Marona Python SDK

Run Agents through one managed API, with structured output, local tools, Hub capabilities, streaming, realtime and A2A collaboration.

Install

Version 2.2.0. Package registry.

Python 3.10+.

python -m pip install --upgrade marona==2.2.0

Set MARONA_API_KEY in your backend environment. Never commit it, log it, or embed a developer key in browser or mobile code. Use an exact accessible marona/* model alias from Marona Platform.

Agent quickstart

Save as examples/agent_example.py. Run without arguments for text, or with a real PNG/JPEG path for document extraction. Use stable application user IDs and a distinct session ID for each conversation.

import asyncio
import base64
import os
import sys
from pathlib import Path

from marona import Agent, Marona, Runner


async def main():
    async with Marona(
        api_key=os.environ["MARONA_API_KEY"],
        base_url=os.getenv("MARONA_RUNTIME_URL", "https://edge.marona.ai"),
    ) as marona:
        agent = Agent(
            name="Document Assistant",
            model=os.getenv("MARONA_MODEL", "marona/qwen-qwen3.8-max-0902"),
            instructions="Answer clearly and concisely.",
            tool_choice="none",
        )
        user_input = "Reply with exactly OK."
        if len(sys.argv) > 1:
            path = Path(sys.argv[1])
            mime = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg"}[path.suffix.lower()]
            data_url = f"data:{mime};base64," + base64.b64encode(path.read_bytes()).decode("ascii")
            user_input = [{"role": "user", "content": [
                {"type": "input_text", "text": "Describe this image and transcribe any visible text."},
                {"type": "input_image", "image_url": data_url},
            ]}]
        result = await Runner.run(
            agent, user_input, user_id="example-user", session_id="example-session"
        )
        print(result.output)


if __name__ == "__main__":
    asyncio.run(main())

python examples/agent_example.py
python examples/agent_example.py image.png

Encode actual file bytes with standard Base64. Image and file blocks accept raw Base64 in image_data / file_data, or complete matching data URLs. Do not encode a path, encode twice, or send placeholder data. Remote URLs must be reachable by the runtime. PDF, audio and realtime support depend on the selected model; render PDF pages to images when native PDF is unavailable.

Structured output through Agent

The following fragment uses the authenticated client from the quickstart. document_input is your text or document message input.

from pydantic import BaseModel, ConfigDict
from marona import Agent, Runner

class Extraction(BaseModel):
    model_config = ConfigDict(extra="forbid")
    text: str

agent = Agent(
    name="Document Reader",
    model="marona/qwen-qwen3.8-max-0902",
    instructions="Extract the document text.",
    output_type=Extraction,
    output_schema_strict=True,
)
result = await Runner.run(agent, document_input, user_id="user-123")
print(result.output.text)

The explicit strict option asks the provider to enforce the supplied schema; the SDK also validates the final result before returning it. Strict mode requires a compatible model and schema. Declare every object property as required, use nullable values for optional fields, and set additionalProperties to false. Unsupported model capabilities or invalid output produce errors; they do not silently fall back to unstructured text. Input schemas, function tool schemas, guardrails, dynamic instructions and lifecycle hooks are also supported.

Apps and user connections

page = await marona.apps.list_async(user_id="user-123", limit=20)
action = await marona.apps.connect_async("calendar", user_id="user-123")
# Complete any returned authorization_url before calling protected tools.
await marona.apps.disconnect_async("calendar")

Listing an App does not authorize its protected operations. Follow the returned connection action for the end user. Use hub.connect to expose a scoped set of App or Skill capabilities to an Agent; Apps authorization and Hub tool discovery are separate operations.

A2A collaboration and serving

import os
from marona import A2APeer, A2AServer, Agent

peer = A2APeer(name="reviewer", url="https://reviewer.example.com")
coordinator = Agent(name="Coordinator", peers=[peer])
server = A2AServer(
    agent=coordinator,
    url="https://coordinator.example.com",
    skills=[{"id": "review", "name": "Review", "description": "Review a document"}],
)
# Apply production authentication and TLS at your service boundary.
server.run(host="127.0.0.1", port=8100)

Peers support Agent Card discovery, REST/JSON-RPC messages, task retrieval, cancellation, continuation and streamed task events. Servers persist task state. Use explicit credentials, restricted skills, HTTPS, and an appropriate durable task directory for production. Local JSON stores are for a single service instance; multi-instance deployments require coordinated task storage.

Push notifications are not implemented.

Brain2Agent (B2A)

Native B2A v0.1 exchanges decoded semantic intent and agent responses. The independent neural service owns signal decoding and feedback encoding, with vendor-hosted MCP, HTTP or local plugin backends. Models and device drivers are supplied by the interface provider; examples use clearly labeled simulation.

Save as examples/b2a_intent.py:

"""Send a simulated semantic intent through native B2A; no raw EEG is inferred.

Set MARONA_API_KEY and optionally MARONA_BASE_URL. A real application obtains
the human's explicit consent and replaces the fixture with its decoder output.
The separate neural service receives response events and encodes feedback.
"""

import asyncio
import json
import os

from marona import Agent, B2AMessage, Marona


async def main() -> None:
    async with Marona(base_url=os.getenv("MARONA_BASE_URL", "https://edge.marona.ai")) as client:
        agent = Agent(
            name="B2A Intent Agent",
            instructions="Interpret the supplied semantic intent in context and respond clearly.",
        )
        session = await client.b2a.open_session_async(
            user_id="example-human",
            device_id="example-neural-interface",
            consent={"granted": True, "scopes": ["intent", "feedback", "context"]},
            capabilities={"input": ["semantic.intent"], "output": ["application/json"]},
            agent=agent,
        )
        try:
            message = B2AMessage.intent(
                session["session_id"],
                session["next_sequence"],
                {
                    "goal": "Explain how an ecosystem maintains balance",
                    "context": {"perspective": "interdependence", "detail": "concise"},
                },
                decoder={"id": "example-fixture", "version": "1"},
                provenance={"simulated": True},
                context_revision=session["context_revision"],
            )
            # Preserve this envelope unchanged when retrying after transport loss.
            exchange = await client.b2a.send_message_async(session["session_id"], message)
            for event in exchange["events"]:
                if event["type"] == "agent.response":
                    # Pass this payload to the independently deployed encoder.
                    print(json.dumps(event["payload"], ensure_ascii=False, indent=2))
        finally:
            await client.b2a.close_session_async(session["session_id"])


if __name__ == "__main__":
    asyncio.run(main())

Keep the same message ID, sequence, timestamp and payload when retrying an uncertain delivery. Read session.next_sequence for the next new message. Inspect error events and explicit cancellation outcomes. Feedback presentation is separate from human confirmation. B2A v0.1 uses HTTP JSON and retained cursor events; it does not claim WebSocket streaming or direct neural-model inference. See the repository's docs/b2a-protocol.md for the complete wire contract.

Typed neural service client

Install marona[b2a] for MCP transport; HTTP works with the base package. B2ACodec calls an independent service and returns validated typed results. The SDK preserves decoder confidence, provenance, uncertainty, and signal identity. Abstention returns intent=None. Simulation is rejected unless explicitly enabled.

from marona import B2ACodec, B2AMessage

codec = B2ACodec(codec_url, token=neural_token, transport="mcp")
decoded = await codec.decode(signal, consent=True)
if decoded.intent is not None:
    message = B2AMessage.intent(session_id, next_sequence, decoded.intent)
    exchange = await client.b2a.send_message_async(session_id, message)

feedback = await codec.encode(agent_response, profile=output_profile, consent=True)
# Pass feedback to the actual interface driver; encoding is not delivery.

Standard operations default to MCP tools decode/encode or HTTP paths /v1/decode//v1/encode. Override decode_tool, encode_tool, decode_path, and encode_path for another provider. Alternate names retain the same contract: a decode result contains status, semantic intent, confidence, uncertainty, and provenance; encoded feedback contains its ID, profile, media type, content, and provenance. Model implementations and weights remain with the provider.

The codec has a bounded timeout, never automatically retries, and requires its own credential and explicit consent on every operation. Agent identity, instructions, and Hub tool scope are independent of codec configuration.

Execution and errors

Use Runner.stream for streamed Agent events and RealtimeRunner for a live session. Delegation and handoffs use explicit Agent graphs. Set execution limits and validate permissions before tools with side effects; cancellation cannot roll back an operation that has already executed.

Inspect the canonical error code, HTTP status, request ID and retry metadata. Correct invalid input and authorization errors before retrying. Use bounded retries for transient failures. Never include credentials or document content in error reports.

See Platform documentation for language specific examples and migration guidance. These files are generated from docs/client-sdk-guides.json; verify them with python scripts/sync_client_sdk_docs.py --check.