LangChain and Provider SDKs

# /// script
# requires-python = ">=3.12,<3.13"
# dependencies = ["langchain-core>=1.2.28,<2"]
# ///
"""Invoke a real LangChain tool against your Chio-governed notes application."""
from langchain_core.tools import StructuredTool
from inputs import note_arguments, save_notes, show_saved
from notes_session import NotesSession


def main():
    texts = note_arguments()
    with NotesSession() as session:
        tool = StructuredTool.from_function(
            func=session.save, name="save_note",
            description="Save a note under this session's Chio write allowance.",
        )
        save_notes(lambda text: tool.invoke({"text": text}), texts)
        show_saved(session)


if __name__ == "__main__":
    main()

# /// script
# requires-python = ">=3.12,<3.13"
# dependencies = ["anthropic>=0.84,<1"]
# ///
"""Connect Anthropic Messages to your Chio-governed notes application."""
import argparse
import json
import os
from anthropic import Anthropic
from notes_session import NotesSession


def run(client, session, model, prompt):
    tools = [{"name": tool["name"],
              "description": tool.get("description", ""),
              "input_schema": tool["inputSchema"]} for tool in session.tools()]
    messages = [{"role": "user", "content": prompt}]
    for _ in range(8):
        response = client.messages.create(
            model=model, messages=messages, tools=tools, max_tokens=512,
        )
        print(json.dumps({"provider_response_id": response.id, "model": response.model, "usage": response.usage.model_dump() if response.usage else None}))
        calls = [block for block in response.content if block.type == "tool_use"]
        if not calls:
            text = "\n".join(block.text for block in response.content if block.type == "text")
            print(text or "The model returned no text or tool call.")
            return
        if len(calls) > 16:
            raise RuntimeError("The model returned more than 16 tool calls in one turn")
        messages.append({"role": "assistant", "content": [
            block.model_dump(exclude_none=True) for block in response.content]})
        results = []
        for call in calls:
            if isinstance(call.input, dict):
                result = session.call(call.name, call.input)
            else:
                result = {"isError": True, "content": [{"type": "text",
                          "text": "Tool arguments must be a JSON object"}]}
            print(json.dumps({"tool": call.name, "result": result}))
            results.append({"type": "tool_result", "tool_use_id": call.id,
                            "content": session.text(result),
                            "is_error": bool(result.get("isError"))})
        messages.append({"role": "user", "content": results})
    raise RuntimeError("Stopped after eight model turns; session evidence is retained")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--model", required=True, help="A tool-capable model available to your account")
    parser.add_argument("prompt", help="Tell the model what notes to save or read")
    args = parser.parse_args()
    if not (os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("ANTHROPIC_AUTH_TOKEN")):
        parser.error("Set ANTHROPIC_API_KEY or ANTHROPIC_AUTH_TOKEN in your environment")
    with Anthropic(timeout=45, max_retries=0) as client, NotesSession() as session:
        run(client, session, args.model, args.prompt)


if __name__ == "__main__":
    main()

# /// script
# requires-python = ">=3.12,<3.13"
# dependencies = ["openai>=2,<3"]
# ///
"""Connect OpenAI Chat Completions to your Chio-governed notes application."""
import argparse
import json
import os
from openai import OpenAI
from notes_session import NotesSession


def run(client, session, model, prompt):
    tools = [{"type": "function", "function": {
        "name": tool["name"], "description": tool.get("description", ""),
        "parameters": tool["inputSchema"],
    }} for tool in session.tools()]
    messages = [{"role": "user", "content": prompt}]
    for _ in range(8):
        response = client.chat.completions.create(
            model=model, messages=messages, tools=tools,
            max_completion_tokens=512,
        )
        print(json.dumps({"provider_response_id": response.id, "model": response.model, "usage": response.usage.model_dump() if response.usage else None}))
        message = response.choices[0].message
        calls = message.tool_calls or []
        if not calls:
            print(message.content or "The model returned no text or tool call.")
            return
        if len(calls) > 16:
            raise RuntimeError("The model returned more than 16 tool calls in one turn")
        messages.append(message.model_dump(exclude_none=True))
        for call in calls:
            if call.type != "function":
                raise RuntimeError("This example accepts function tool calls only")
            try:
                arguments = json.loads(call.function.arguments)
                if not isinstance(arguments, dict):
                    raise ValueError("Tool arguments must be a JSON object")
            except (json.JSONDecodeError, ValueError) as error:
                result = {"isError": True, "content": [{"type": "text", "text": str(error)}]}
            else:
                result = session.call(call.function.name, arguments)
            print(json.dumps({"tool": call.function.name, "result": result}))
            messages.append({"role": "tool", "tool_call_id": call.id,
                             "content": json.dumps(result)})
    raise RuntimeError("Stopped after eight model turns; session evidence is retained")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--model", required=True, help="A tool-capable model available to your account")
    parser.add_argument("prompt", help="Tell the model what notes to save or read")
    args = parser.parse_args()
    if not os.environ.get("OPENAI_API_KEY"):
        parser.error("Set OPENAI_API_KEY in your environment")
    with OpenAI(timeout=45, max_retries=0) as client, NotesSession() as session:
        run(client, session, args.model, args.prompt)


if __name__ == "__main__":
    main()

{
  "id": "0092d85416b8bce76a842e988f79566ab266dd900ac119d41a134ff8649d5a5e",
  "timestamp": 1788950127,
  "capability_id": "cap-01a085bc-9f6f-7610-b34f-68e958ef2a9b",
  "tool_server": "notes",
  "tool_name": "createNote",
  "action": {
    "parameters": {
      "body": {
        "text": "This exceeds the write grant"
      }
    },
    "parameter_hash": "85ba62e99b5f65a6b6d514f59ffc86154151ad54a86f72297968bfbd81a89b56"
  },
  "decision": {
    "verdict": "deny",
    "reason": "invocation budget exhausted for capability cap-01a085bc-9f6f-7610-b34f-68e958ef2a9b",
    "guard": "kernel"
  },
  "receipt_kind": "mediated_decision",
  "boundary_class": "prevent",
  "tool_origin": "caller_executed",
  "redaction_mode": "none",
  "content_hash": "74234e98afe7498fb5daf1f36ac2d78acc339464f950703b8c019892f982b90b",
  "policy_hash": "d2ed7d01074c952ab46491c89a53e51d03d9e089fa670861788ace0c28e42fcb",
  "metadata": {
    "attribution": {
      "delegation_depth": 0,
      "issuer_key": "ba6bdf961aec7c011b32cdfd2b4a72beeb310a1677dc38f25f3849c8be4cf341",
      "subject_key": "e9256adf30c15014b2474b0f22288478ac1d311cda212d0d2328be5af1c4f6a1"
    },
    "budget_authority": {
      "authority_profile": "authoritative_hold_event",
      "guarantee_level": "single_node_atomic",
      "metering_profile": "max_cost_preauthorize_then_reconcile_actual"
    },
    "chio_receipt_signing_nonce": "rcpt-01a085bc-a213-7572-8cc7-f33dc6c878fc",
    "receipt_context": {
      "request_id": "mcp-edge-req-f5c381fb5ee94535995a236cac6600be"
    }
  },
  "trust_level": "mediated",
  "kernel_key": "ba6bdf961aec7c011b32cdfd2b4a72beeb310a1677dc38f25f3849c8be4cf341",
  "signature": "03a24bc78b1cd48b1830130d64e1fb3d11a0669d734374336518ff87532498838feab60ebe94cb9c8585a0f38266da83de129ebfd6343a71ce039776eaf77503"
}