Agent Framework SDKs

# /// script
# requires-python = ">=3.12,<3.13"
# dependencies = ["chio-crewai", "crewai>=1.15,<2"]
# [tool.uv.sources]
# chio-adapter-base = { path = "sdk/chio-adapter-base" }
# chio-sdk-python = { path = "sdk/chio-sdk-python" }
# chio-crewai = { path = "sdk/chio-crewai" }
# chio-autogen = { path = "sdk/chio-autogen" }
# chio-llamaindex = { path = "sdk/chio-llamaindex" }
# ///
"""Use Chio's supported CrewAI wrapper with an existing MCP session."""
import os
os.environ.setdefault("OTEL_SDK_DISABLED", "true")
os.environ.setdefault("CREWAI_TELEMETRY_DISABLED", "true")
from chio_crewai import ChioBaseTool
from inputs import note_arguments, save_notes, show_saved
from notes_session import NotesSession
from tool_inputs import CreateNoteInput

def main():
    with NotesSession() as session:
        tool = ChioBaseTool.from_mcp(session, tool_name="createNote", args_schema=CreateNoteInput,
                                    description="Save a note using this session's Chio write grant.")
        save_notes(lambda text: tool.run(body={"text": text}), note_arguments())
        show_saved(session)

if __name__ == "__main__": main()

# /// script
# requires-python = ">=3.12,<3.13"
# dependencies = ["chio-adapter-base", "autogen-agentchat>=0.7,<0.8", "autogen-ext[openai]>=0.7,<0.8"]
# [tool.uv.sources]
# chio-adapter-base = { path = "sdk/chio-adapter-base" }
# chio-sdk-python = { path = "sdk/chio-sdk-python" }
# ///
"""A writer and an auditor use separate Chio authority in modern AgentChat."""
import argparse
import asyncio
import json
import os
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from chio_adapter_base.agentchat import mcp_tool
from chio_adapter_base.mcp import McpToolBinding
from pydantic import BaseModel, ConfigDict
from notes_session import NotesSession
from tool_inputs import CreateNoteInput


class ListNotesInput(BaseModel):
    model_config = ConfigDict(extra="forbid")


def agent(name, session, client):
    return AssistantAgent(
        name, model_client=client,
        tools=[
            mcp_tool(McpToolBinding(session, "createNote"), schema=CreateNoteInput,
                     description="Attempt to save a note. Chio decides whether this session may write."),
            mcp_tool(McpToolBinding(session, "listNotes"), schema=ListNotesInput,
                     description="Read the notes actually saved by the application."),
        ],
        system_message="Use tools for facts about saved notes. Report refusals accurately; never claim a failed write succeeded. Do not retry a refused or uncertain write.",
        max_tool_iterations=6, reflect_on_tool_use=True,
    )


async def main(args):
    configuration = dict(model=args.model, api_key=os.environ["OPENAI_API_KEY"],
                         timeout=45, max_retries=0, parallel_tool_calls=False,
                         max_tokens=700)
    if os.environ.get("OPENAI_BASE_URL"):
        configuration.update(base_url=os.environ["OPENAI_BASE_URL"], model_info={
            "vision": False, "function_calling": True, "json_output": False,
            "family": "unknown", "structured_output": False,
        })
    client = OpenAIChatCompletionClient(**configuration)
    try:
        with NotesSession(role="writer") as writer_session:
            writer = agent("writer", writer_session, client)
            result = await writer.run(task=args.prompt)
            records = [message.model_dump(mode="json") for message in result.messages]
            (writer_session.directory / "agentchat.json").write_text(json.dumps(records, indent=2))
            print(json.dumps({"agent": "writer", "messages": records}))
            handoff = str(result.messages[-1].content)
        with NotesSession(role="auditor") as auditor_session:
            auditor = agent("auditor", auditor_session, client)
            task = "Read the saved notes and audit this writer handoff against the actual data: " + handoff
            if args.test_boundary:
                task += " First attempt createNote once with text 'Auditor boundary probe'; this is an explicit refusal test. Then listNotes and report whether the attempted write was refused. Do not retry it."
            result = await auditor.run(task=task)
            records = [message.model_dump(mode="json") for message in result.messages]
            (auditor_session.directory / "agentchat.json").write_text(json.dumps(records, indent=2))
            print(json.dumps({"agent": "auditor", "messages": records}))
    finally:
        await client.close()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--model", required=True)
    parser.add_argument("--test-boundary", action="store_true", help="Ask the auditor to attempt one unauthorized write")
    parser.add_argument("prompt", help="The writer's task; the auditor checks its saved result")
    args = parser.parse_args()
    if not os.environ.get("OPENAI_API_KEY"): parser.error("Set OPENAI_API_KEY in your environment")
    asyncio.run(main(args))

# /// script
# requires-python = ">=3.12,<3.13"
# dependencies = ["chio-autogen", "pyautogen>=0.2,<0.3", "numpy<2"]
# [tool.uv.sources]
# chio-adapter-base = { path = "sdk/chio-adapter-base" }
# chio-sdk-python = { path = "sdk/chio-sdk-python" }
# chio-crewai = { path = "sdk/chio-crewai" }
# chio-autogen = { path = "sdk/chio-autogen" }
# chio-llamaindex = { path = "sdk/chio-llamaindex" }
# ///
"""Register remote Chio execution with classic AutoGen's actual tool registry."""
import json
from autogen import ConversableAgent
from chio_adapter_base.mcp import McpToolBinding, McpToolError
from chio_autogen import ChioFunctionRegistry
from inputs import note_arguments, save_notes, show_saved
from notes_session import NotesSession
from tool_inputs import CreateNoteInput

def main():
    with NotesSession() as session:
        executor = ConversableAgent("notes_executor", llm_config=False, human_input_mode="NEVER", code_execution_config=False)
        binding = McpToolBinding(session, "createNote")
        registry = ChioFunctionRegistry(agent=executor, chio_client=None, server_id="notes")
        registry.register_mcp(binding, schema=CreateNoteInput, description="Save a note with Chio.", name="save_note")
        def save(text):
            succeeded, result = executor.execute_function({"name": "save_note", "arguments": json.dumps({"body": {"text": text}})})
            if not succeeded:
                if binding.last_execution and binding.last_execution.is_error:
                    raise McpToolError(binding.last_execution)
                raise RuntimeError(result["content"])
            return result["content"]
        save_notes(save, note_arguments())
        show_saved(session)

if __name__ == "__main__": main()

# /// script
# requires-python = ">=3.12,<3.13"
# dependencies = ["chio-llamaindex", "llama-index-core>=0.14,<1"]
# [tool.uv.sources]
# chio-adapter-base = { path = "sdk/chio-adapter-base" }
# chio-sdk-python = { path = "sdk/chio-sdk-python" }
# chio-crewai = { path = "sdk/chio-crewai" }
# chio-autogen = { path = "sdk/chio-autogen" }
# chio-llamaindex = { path = "sdk/chio-llamaindex" }
# ///
"""Use Chio's supported LlamaIndex wrapper for remote tool execution."""
from chio_adapter_base.mcp import McpToolBinding, McpToolError
from chio_llamaindex import ChioFunctionTool
from inputs import note_arguments, save_notes, show_saved
from notes_session import NotesSession
from tool_inputs import CreateNoteInput

def main():
    with NotesSession() as session:
        binding = McpToolBinding(session, "createNote")
        tool = ChioFunctionTool.from_mcp(binding, fn_schema=CreateNoteInput,
                                       description="Save a note under this session's Chio write grant.")
        def save(text):
            output = tool.call(body={"text": text})
            if output.is_error:
                if binding.last_execution and binding.last_execution.is_error:
                    raise McpToolError(binding.last_execution)
                raise RuntimeError(output.content)
            return output.content
        save_notes(save, note_arguments())
        show_saved(session)

if __name__ == "__main__": main()

# /// script
# requires-python = ">=3.11"
# dependencies = ["chio-sdk"]
# [tool.uv.sources]
# chio-sdk = { path = "sdk/chio-py" }
# ///
"""Verify a session's signed receipts and match them to its MCP calls/results."""
import argparse
import json
from pathlib import Path
import sqlite3
from chio.invariants import verify_receipt_with_trusted_signers, canonicalize_json, sha256_hex_utf8


def verify(directory, trusted_key):
    directory = Path(directory)
    wire = [json.loads(line) for line in (directory/'mcp.ndjson').read_text().splitlines() if line.strip()]
    requests = {row['message']['id']: row['message'] for row in wire
                if row['direction'] == 'client' and row['message'].get('method') == 'tools/call'}
    connection = sqlite3.connect((directory/'receipts.db').resolve().as_uri()+'?mode=ro', uri=True)
    try:
        receipts = {row[0]: json.loads(row[1]) for row in connection.execute('SELECT receipt_id, raw_json FROM chio_tool_receipts')}
    finally: connection.close()
    calls = []; seen = set()
    for row in wire:
        response = row['message']
        if row['direction'] != 'server' or response.get('id') not in requests: continue
        request = requests[response['id']]
        result = response.get('result', {})
        association = result.get('_meta', {}).get('chioReceipt', {})
        receipt = receipts[association['receiptId']]
        checks = verify_receipt_with_trusted_signers(receipt, [trusted_key])
        if not checks['ok']: raise ValueError(f'Invalid or untrusted receipt {receipt["id"]}: {checks}')
        if receipt['id'] in seen: raise ValueError('A receipt was reused across distinct MCP calls')
        seen.add(receipt['id'])
        if receipt['metadata']['receipt_context']['request_id'] != association['requestId']:
            raise ValueError('Result and signed receipt refer to different operations')
        if receipt['action']['parameters'] != request['params']['arguments'] or receipt['tool_name'] != request['params']['name']:
            raise ValueError('Signed receipt does not match the submitted tool and arguments')
        allowed = receipt['decision']['verdict'] == 'allow'
        failed = result.get('isError', False)
        if allowed:
            mapped = {key: result[key] for key in ('content', 'structuredContent', 'isError') if key in result}
            mapped.setdefault('content', [])
            mapped.setdefault('isError', False)
            if sha256_hex_utf8(canonicalize_json(mapped)) != receipt['content_hash']:
                raise ValueError('The returned tool result differs from its signed output hash')
        if not failed and not allowed: raise ValueError('A denied operation was reported as successful')
        calls.append({'tool': request['params']['name'], 'arguments': request['params']['arguments'],
                      'outcome': 'denied' if not allowed else ('execution_failed' if failed else 'completed'),
                      'receipt_id': receipt['id'], 'request_id': association['requestId'],
                      'verification': checks, 'result': result, 'receipt': receipt})
    if len(calls) != len(requests) or not calls:
        raise ValueError('At least one tool call has no verifiable terminal receipt')
    return {'verified': True, 'calls': calls, 'policy': json.loads((directory/'authority.json').read_text()),
            'role': (directory/'role.txt').read_text(), 'trusted_kernel_key': trusted_key}


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('session', type=Path)
    parser.add_argument('--trusted-key', type=Path, required=True, help='An independently selected kernel public-key file')
    args = parser.parse_args()
    print(json.dumps(verify(args.session, args.trusted_key.read_text().strip()), indent=2))