Recipes.

import asyncio
from pathlib import Path

from matilda_agent_sdk import MatildaClient, Runner, create_session
from matilda_client import create_file_token_store


async def main():
    token_store = create_file_token_store(Path.home() / ".matilda" / "tokens.json")

    runner = Runner(MatildaClient(base_url="https://matilda.maincode.com/api"))

    # Try to restore persisted tokens, fall back to interactive login
    restored = await runner.auth.restore(
        client_id="matilda-code",
        token_store=token_store.store,
        token_lock=token_store.lock,
    )
    if not restored:
        print("Starting device flow authentication...")
        await runner.auth.login_with_device_flow(
            client_id="matilda-code",
            token_store=token_store.store,
            token_lock=token_store.lock,
        )
        print("Authenticated!")

    session = create_session(
        {"name": "cli-assistant", "purpose": "general", "instructions": "Be helpful, concise, and friendly."},
        runner=runner,
    )

    while True:
        user_input = await asyncio.to_thread(input, "\nYou: ")
        if not user_input.strip() or user_input.lower() == "exit":
            break

        print("Agent: ", end="", flush=True)
        async for event in session.stream(user_input):
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
        print()


asyncio.run(main())

import asyncio
import json

from matilda_agent_sdk import MatildaClient, Runner, ToolResult


async def get_weather(args, ctx):
    city = args.get("city", "unknown")
    # In reality, call a weather API
    return ToolResult(content=json.dumps({"city": city, "temp": 22, "condition": "sunny"}))


async def calculate(args, ctx):
    try:
        # NOTE: eval is a stand-in for a real expression parser — never eval
        # untrusted input in production.
        result = eval(str(args["expression"]), {"__builtins__": {}}, {})
        return ToolResult(content=str(result))
    except Exception:
        return ToolResult(content="Invalid expression", is_error=True)


client_tools = [
    {"name": "get_weather", "description": "Get current weather for a city", "parameters": {"type": "object"}},
    {"name": "calculate", "description": "Evaluate a math expression", "parameters": {"type": "object"}},
]

tool_handlers = {"get_weather": get_weather, "calculate": calculate}


async def main():
    runner = Runner(MatildaClient(base_url="https://matilda.maincode.com/api"))

    if not await runner.auth.get_tokens():
        await runner.auth.login_with_device_flow(client_id="matilda-code")

    async for event in runner.stream(
        {"name": "assistant", "instructions": "Use the available tools to answer."},
        "What is the weather in Sydney, and what is 15 * 23?",
        tool_handlers=tool_handlers,
        client_tools=client_tools,
    ):
        if event.type == "client.tool.requested":
            print(f"→ {event.name}({json.dumps(event.args)})")
        if event.type == "client.tool.result":
            print(f"← {event.result}")
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)


asyncio.run(main())

import asyncio

from matilda_agent_sdk import Agent, MatildaClient, Runner


async def main():
    runner = Runner(MatildaClient(base_url="https://matilda.maincode.com/api"))

    if not await runner.auth.get_tokens():
        await runner.auth.login_with_device_flow(client_id="matilda-code")

    security = Agent(
        name="security",
        purpose="analysis",
        instructions="Review for vulnerabilities. Be specific.",
    )

    performance = Agent(
        name="performance",
        purpose="analysis",
        instructions="Review for efficiency. Be specific.",
    )

    synthesiser = Agent(
        name="synthesiser",
        purpose="analysis",
        instructions="Merge reviews into a prioritised action list. Use 🔴 🟡 🟢 priority.",
    )

    code = 'def get_user_data(user_id, db): query = "SELECT * FROM users WHERE id = " + user_id'

    sec, perf = await asyncio.gather(
        runner.run(security, f"Review:\n```python\n{code}\n```"),
        runner.run(performance, f"Review:\n```python\n{code}\n```"),
    )

    combined = f"## Security\n{sec.final_output}\n\n## Performance\n{perf.final_output}"
    result = await runner.run(synthesiser, combined)
    print(result.final_output)


asyncio.run(main())

import asyncio

from matilda_agent_sdk import Agent, run


def reviewer_instructions(ctx):
    lang = ctx["metadata"].get("language", "auto-detect")
    strictness = ctx["metadata"].get("strictness", "normal")
    return "\n".join(
        [
            "Review the following code.",
            f"Language: {lang}",
            f"Strictness: {strictness}",
            "Focus on: correctness, security, and readability.",
            "Cite line numbers when possible.",
        ]
    )


agent = Agent(name="code-reviewer", purpose="code", instructions=reviewer_instructions)


async def main():
    result = await run(
        agent,
        "def add(a, b): return a + b",
        metadata={"language": "Python", "strictness": "strict"},
    )

    print(result.final_output)


asyncio.run(main())

import asyncio

from matilda_agent_sdk import resume_agent_stream, stream


async def main():
    stream_id = None
    last_event_id = None
    received_text = ""

    print("Starting stream...")
    try:
        async for event in stream({"name": "writer"}, "Write a very long essay about Australia."):
            if event.type == "stream.started":
                stream_id = event.stream_id
            if event.type == "response.cursor":
                last_event_id = event.last_event_id
            if event.type == "response.output_text.delta":
                received_text += event.delta
                # Simulate disconnect after 500 chars
                if len(received_text) > 500:
                    print("\n--- Simulated disconnect ---")
                    break
    except Exception as err:
        print(f"Disconnected: {err}")

    print(f"Received {len(received_text)} chars before disconnect.")

    # Resume from the last cursor
    if stream_id:
        print("\n--- Resuming ---")
        result = await resume_agent_stream(
            stream_id,
            last_event_id,
            on_event=lambda event: (
                print(event.delta, end="", flush=True)
                if event.type == "response.output_text.delta"
                else None
            ),
        )
        print(f"\nTotal output: {len(result.final_output)} chars")


asyncio.run(main())

import asyncio
from pathlib import Path

from matilda_agent_sdk import MatildaClient, Runner
from matilda_client import create_file_token_store


async def main():
    token_path = Path.home() / ".matilda" / "tokens.json"
    token_store = create_file_token_store(token_path)

    runner = Runner(MatildaClient(base_url="https://matilda.maincode.com/api"))

    # Restore persisted tokens or login interactively
    restored = await runner.auth.restore(
        client_id="matilda-code",
        token_store=token_store.store,
        token_lock=token_store.lock,
    )
    if not restored:
        await runner.auth.login_with_device_flow(
            client_id="matilda-code",
            token_store=token_store.store,
            token_lock=token_store.lock,
        )

    # Runner is ready — tokens auto-refresh on 401
    result = await runner.run(
        {"name": "helper", "instructions": "Be concise."},
        "What is the capital of Australia?",
    )
    print(result.final_output)

    # Later: logout clears the token store
    # await runner.auth.logout()


asyncio.run(main())