~/selectools/examples zsh

Selectools examples — 115 runnable Python scripts

115 runnable scripts covering agents, RAG, multi-agent graphs, evals, streaming, and guardrails. 44 run without an API key.

# 115 files match
\n\"\"\"\n )\n\ndocs = DocumentLoader.from_html(html_path)\nprint(f\"\\nHTML: {len(docs)} documents\")\nprint(f\" Text: {docs[0].text[:80]}...\")\n\n# 4. URL Loader (would fetch from web -- shown as pattern)\nprint(\"\\nURL loader pattern:\")\nprint(\" docs = DocumentLoader.from_url('https://example.com/article')\")\nprint(\" # Auto-detects content type, strips HTML tags\")\n\n# Cleanup\nfor p in [csv_path, json_path, html_path]:\n os.unlink(p)\n\nprint(\"\\nDone!\")\n", "81_multimodal_messages.py": "#!/usr/bin/env python3\n\"\"\"\nMultimodal Messages -- send images to your agent.\n\nNo API key needed for this demo. Shows how to create messages with images\nfor vision-enabled models (GPT-4o, Claude 3.5, Gemini).\n\nRun: python examples/81_multimodal_messages.py\n\"\"\"\n\nfrom selectools.types import ContentPart, Message, Role, image_message, text_content\n\nprint(\"=== Multimodal Messages Example ===\\n\")\n\n# 1. Simple image message from URL\nmsg = image_message(\"https://example.com/photo.jpg\", \"What do you see in this image?\")\nprint(f\"Image URL message: {len(msg.content_parts)} parts\")\nprint(f\" Text: {msg.content_parts[0].text!r}\")\nprint(f\" Image: {msg.content_parts[1].image_url!r}\")\n\n# 2. Multiple images in one message\nmsg_multi = Message(\n role=Role.USER,\n content=\"Compare these two images\",\n content_parts=[\n ContentPart(type=\"text\", text=\"Compare these two product photos\"),\n ContentPart(type=\"image_url\", image_url=\"https://example.com/product_a.jpg\"),\n ContentPart(type=\"image_url\", image_url=\"https://example.com/product_b.jpg\"),\n ],\n)\nprint(f\"\\nMulti-image message: {len(msg_multi.content_parts)} parts\")\n\n# 3. Extract text from multimodal message\nextracted = text_content(msg_multi)\nprint(f\"Extracted text: {extracted!r}\")\n\n# 4. Backward compatibility -- str content still works\nplain = Message(role=Role.USER, content=\"Just plain text, no images\")\nprint(f\"\\nPlain text: {text_content(plain)!r}\")\nprint(f\"content_parts is None: {plain.content_parts is None}\")\n\n# 5. Usage with an agent (pattern)\nprint(\n \"\"\"\n# With a vision-enabled model:\nfrom selectools import Agent\nfrom selectools.providers import OpenAIProvider\n\nagent = Agent(tools=[], provider=OpenAIProvider())\nmsg = image_message(\"photo.jpg\", \"Describe this image\")\nresult = agent.run(msg)\nprint(result.content)\n\"\"\"\n)\n\nprint(\"Done!\")\n", "82_code_execution.py": "#!/usr/bin/env python3\n\"\"\"\nCode Execution Tools -- run Python and shell commands from agents.\n\nNo API key needed. Demonstrates the execute_python and execute_shell tools.\n\nWARNING: These tools execute code on your local machine. Do not use with\nuntrusted input without sandboxing.\n\nRun: python examples/82_code_execution.py\n\"\"\"\n\nfrom selectools.toolbox.code_tools import execute_python, execute_shell\n\nprint(\"=== Code Execution Tools Example ===\\n\")\n\n# 1. Execute Python code\nprint(\"--- Python Execution ---\")\nresult = execute_python.function(\"import math; print(f'Pi = {math.pi:.6f}')\")\nprint(f\"Result: {result}\")\n\n# 2. Multi-line Python\nresult = execute_python.function(\n \"\"\"\ndata = [1, 2, 3, 4, 5]\ntotal = sum(data)\navg = total / len(data)\nprint(f\"Sum: {total}, Average: {avg}\")\n\"\"\"\n)\nprint(f\"Multi-line: {result}\")\n\n# 3. Shell commands\nprint(\"--- Shell Execution ---\")\nresult = execute_shell.function(\"echo 'Hello from shell' && date\")\nprint(f\"Shell: {result}\")\n\n# 4. With timeout\nresult = execute_python.function(\"print('fast!')\", timeout=5)\nprint(f\"With timeout: {result}\")\n\n# 5. Error handling\nresult = execute_python.function(\"1/0\")\nprint(f\"Error output: {result[:100]}\")\n\n# 6. Agent integration pattern\nprint(\n \"\"\"\n--- Agent Pattern ---\nfrom selectools import Agent\nfrom selectools.providers import OpenAIProvider\nfrom selectools.toolbox.code_tools import execute_python\n\nagent = Agent(\n tools=[execute_python],\n provider=OpenAIProvider(),\n)\nresult = agent.run(\"Calculate the first 10 Fibonacci numbers\")\n# Agent writes and executes Python code to solve the task\n\"\"\"\n)\n\nprint(\"Done!\")\n", "83_web_search.py": "#!/usr/bin/env python3\n\"\"\"\nWeb Search Tools -- search the web and scrape URLs.\n\nNo API key needed. Uses DuckDuckGo for search (no rate limits for moderate use).\n\nRun: python examples/83_web_search.py\n\"\"\"\n\nfrom selectools.toolbox.search_tools import scrape_url, web_search\n\nprint(\"=== Web Search Tools Example ===\\n\")\n\n# Note: These tools make real HTTP requests.\n# Uncomment to test with live web access:\n\n# 1. Web search (DuckDuckGo)\n# result = web_search.function(\"Python AI agent frameworks 2026\")\n# print(f\"Search results:\\n{result[:500]}\")\n\n# 2. Scrape a URL\n# result = scrape_url.function(\"https://example.com\")\n# print(f\"Scraped:\\n{result[:300]}\")\n\n# Show the API pattern\nprint(\"web_search tool:\")\nprint(f\" Name: {web_search.name}\")\nprint(f\" Description: {web_search.description}\")\nprint(f\" Parameters: query (str), num_results (int, default=5)\")\n\nprint(f\"\\nscrape_url tool:\")\nprint(f\" Name: {scrape_url.name}\")\nprint(f\" Description: {scrape_url.description}\")\nprint(f\" Parameters: url (str), selector (str, optional CSS selector)\")\n\nprint(\n \"\"\"\n--- Agent Pattern ---\nfrom selectools import Agent\nfrom selectools.providers import OpenAIProvider\nfrom selectools.toolbox.search_tools import web_search, scrape_url\n\nagent = Agent(\n tools=[web_search, scrape_url],\n provider=OpenAIProvider(),\n)\nresult = agent.run(\"Search for the latest Python release and summarize\")\n\"\"\"\n)\n\nprint(\"Done!\")\n", "84_github_tools.py": "#!/usr/bin/env python3\n\"\"\"\nGitHub Tools -- search repos, read files, list issues from agents.\n\nNo API key needed (optional GITHUB_TOKEN increases rate limit from 60 to 5000/hr).\nRead-only operations only.\n\nRun: python examples/84_github_tools.py\n\"\"\"\n\nfrom selectools.toolbox.github_tools import github_get_file, github_list_issues, github_search_repos\n\nprint(\"=== GitHub Tools Example ===\\n\")\n\n# Note: These tools make real API calls to GitHub.\n# Uncomment to test:\n\n# 1. Search repositories\n# result = github_search_repos.function(\"python ai agent framework\", max_results=3)\n# print(f\"Repos:\\n{result}\\n\")\n\n# 2. Get a file\n# result = github_get_file.function(\"johnnichev/selectools\", \"README.md\")\n# print(f\"File content:\\n{result[:200]}...\\n\")\n\n# 3. List issues\n# result = github_list_issues.function(\"johnnichev/selectools\", state=\"open\", max_results=5)\n# print(f\"Issues:\\n{result}\\n\")\n\n# Show tool metadata\nfor tool in [github_search_repos, github_get_file, github_list_issues]:\n print(f\"{tool.name}:\")\n print(f\" {tool.description}\")\n print()\n\nprint(\n \"\"\"\n--- Agent Pattern ---\nfrom selectools import Agent\nfrom selectools.providers import OpenAIProvider\nfrom selectools.toolbox.github_tools import github_search_repos, github_get_file\n\nagent = Agent(\n tools=[github_search_repos, github_get_file],\n provider=OpenAIProvider(),\n)\nresult = agent.run(\"Find the top Python AI frameworks and read their README\")\n\"\"\"\n)\n\nprint(\"Set GITHUB_TOKEN env var for higher rate limits (5000/hr vs 60/hr)\")\nprint(\"Done!\")\n", "85_database_query.py": "#!/usr/bin/env python3\n\"\"\"\nDatabase Query Tools -- SQL queries from agents (read-only).\n\nNo API key needed. Creates a sample SQLite database and queries it.\nAlso supports PostgreSQL with psycopg2.\n\nRun: python examples/85_database_query.py\n\"\"\"\n\nimport os\nimport sqlite3\nimport tempfile\n\nfrom selectools.toolbox.db_tools import query_sqlite\n\nprint(\"=== Database Query Tools Example ===\\n\")\n\n# Create a sample database\ndb_path = os.path.join(tempfile.mkdtemp(), \"sample.db\")\nconn = sqlite3.connect(db_path)\nconn.execute(\n \"\"\"CREATE TABLE employees (\n id INTEGER PRIMARY KEY,\n name TEXT,\n department TEXT,\n salary REAL\n)\"\"\"\n)\nconn.executemany(\n \"INSERT INTO employees VALUES (?, ?, ?, ?)\",\n [\n (1, \"Alice\", \"Engineering\", 120000),\n (2, \"Bob\", \"Marketing\", 95000),\n (3, \"Charlie\", \"Engineering\", 115000),\n (4, \"Diana\", \"Sales\", 105000),\n (5, \"Eve\", \"Engineering\", 130000),\n ],\n)\nconn.commit()\nconn.close()\n\n# 1. Basic query\nprint(\"--- All employees ---\")\nresult = query_sqlite.function(db_path, \"SELECT * FROM employees\")\nprint(result)\n\n# 2. Filtered query\nprint(\"\\n--- Engineering team ---\")\nresult = query_sqlite.function(\n db_path, \"SELECT name, salary FROM employees WHERE department = 'Engineering'\"\n)\nprint(result)\n\n# 3. Aggregation\nprint(\"\\n--- Department stats ---\")\nresult = query_sqlite.function(\n db_path,\n \"SELECT department, COUNT(*) as count, AVG(salary) as avg_salary FROM employees GROUP BY department\",\n)\nprint(result)\n\n# Cleanup\nos.unlink(db_path)\n\nprint(\n \"\"\"\n--- PostgreSQL Pattern ---\nfrom selectools.toolbox.db_tools import query_postgres\n\nresult = query_postgres.function(\n \"postgresql://user:pass@localhost:5432/mydb\",\n \"SELECT * FROM users LIMIT 10\"\n)\n\n--- Agent Pattern ---\nfrom selectools import Agent\nfrom selectools.providers import OpenAIProvider\nfrom selectools.toolbox.db_tools import query_sqlite\n\nagent = Agent(tools=[query_sqlite], provider=OpenAIProvider())\nresult = agent.run(\"What's the average salary by department?\")\n# Agent generates SQL and executes it (read-only mode)\n\"\"\"\n)\n\nprint(\"Done!\")\n", "86_azure_openai.py": "#!/usr/bin/env python3\n\"\"\"\nAzure OpenAI Provider -- use OpenAI models via Azure endpoints.\n\nRequires: AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_API_KEY env vars.\nRun: python examples/86_azure_openai.py\n\"\"\"\n\nprint(\"=== Azure OpenAI Provider Example ===\\n\")\n\nprint(\n \"\"\"\nfrom selectools import Agent, AgentConfig\nfrom selectools.providers import AzureOpenAIProvider\n\n# Option 1: Explicit configuration\nprovider = AzureOpenAIProvider(\n azure_endpoint=\"https://my-resource.openai.azure.com\",\n api_key=\"your-azure-api-key\",\n azure_deployment=\"gpt-4o\", # Your deployment name\n api_version=\"2024-10-21\",\n)\n\n# Option 2: Environment variables\n# Set AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_API_KEY\nprovider = AzureOpenAIProvider(azure_deployment=\"gpt-4o\")\n\n# Option 3: Azure AD authentication (no API key needed)\nprovider = AzureOpenAIProvider(\n azure_endpoint=\"https://my-resource.openai.azure.com\",\n azure_ad_token=\"your-aad-token\",\n azure_deployment=\"gpt-4o\",\n)\n\n# Use like any other provider\nagent = Agent(\n tools=[],\n provider=provider,\n config=AgentConfig(model=\"gpt-4o\"),\n)\nresult = agent.run(\"Hello from Azure!\")\nprint(result.content)\n\n# Supports all features: streaming, tool calling, structured output\nasync for chunk in agent.astream(\"Stream from Azure\"):\n print(chunk.content, end=\"\")\n\"\"\"\n)\n\nprint(\"Set AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_API_KEY to use.\")\nprint(\"Done!\")\n", "87_otel_observer.py": "#!/usr/bin/env python3\n\"\"\"\nOpenTelemetry Observer -- send agent traces to Datadog, Jaeger, Grafana.\n\nMaps selectools observer events to OTel GenAI semantic convention spans.\nWorks with any OTel-compatible backend.\n\nPrerequisites: pip install opentelemetry-api opentelemetry-sdk\nRun: python examples/87_otel_observer.py\n\"\"\"\n\nprint(\"=== OpenTelemetry Observer Example ===\\n\")\n\nprint(\n \"\"\"\nfrom selectools import Agent, AgentConfig\nfrom selectools.providers import OpenAIProvider\nfrom selectools.observe.otel import OTelObserver\n\n# Create the observer\notel = OTelObserver(tracer_name=\"my-agent-service\")\n\n# Attach to your agent\nagent = Agent(\n tools=[...],\n provider=OpenAIProvider(),\n config=AgentConfig(\n model=\"gpt-4o\",\n observers=[otel], # Traces flow to OTel\n ),\n)\n\n# Run as normal -- spans are created automatically\nresult = agent.run(\"Search and summarize\")\n\n# Spans created:\n# - agent.run (root span)\n# - gen_ai.chat (LLM call, with model + token counts)\n# - tool.execute (tool call, with name + duration)\n# - gen_ai.chat (second LLM call)\n\n# Configure your exporter (Jaeger, OTLP, etc.):\nfrom opentelemetry.sdk.trace import TracerProvider\nfrom opentelemetry.sdk.trace.export import BatchSpanProcessor\nfrom opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter\n\nprovider = TracerProvider()\nprovider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))\n\n# Traces appear in Datadog, Grafana, Jaeger, etc.\n\"\"\"\n)\n\nprint(\"Install: pip install opentelemetry-api opentelemetry-sdk\")\nprint(\"Done!\")\n", "88_langfuse_observer.py": "#!/usr/bin/env python3\n\"\"\"\nLangfuse Observer -- send agent traces to Langfuse for LLM observability.\n\nLangfuse is the most popular open-source LLM observability platform.\nTraces include LLM calls, tool executions, costs, and latencies.\n\nPrerequisites: pip install langfuse\nRun: python examples/88_langfuse_observer.py\n\"\"\"\n\nprint(\"=== Langfuse Observer Example ===\\n\")\n\nprint(\n \"\"\"\nfrom selectools import Agent, AgentConfig\nfrom selectools.providers import OpenAIProvider\nfrom selectools.observe.langfuse import LangfuseObserver\n\n# Option 1: Explicit keys\nlangfuse = LangfuseObserver(\n public_key=\"pk-...\",\n secret_key=\"sk-...\",\n host=\"https://cloud.langfuse.com\", # or self-hosted URL\n)\n\n# Option 2: Environment variables\n# Set LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_HOST\nlangfuse = LangfuseObserver()\n\n# Attach to your agent\nagent = Agent(\n tools=[...],\n provider=OpenAIProvider(),\n config=AgentConfig(\n model=\"gpt-4o\",\n observers=[langfuse],\n ),\n)\n\n# Run as normal -- traces sent to Langfuse automatically\nresult = agent.run(\"Analyze this data\")\n\n# Langfuse dashboard shows:\n# - Trace timeline with LLM calls and tool executions\n# - Token counts and costs per call\n# - Model used, latency, input/output preview\n# - Error tracking and debugging\n\n# Flush on shutdown\nlangfuse.shutdown()\n\"\"\"\n)\n\nprint(\"Install: pip install langfuse\")\nprint(\"Dashboard: https://cloud.langfuse.com (free tier available)\")\nprint(\"Self-hosted: https://langfuse.com/docs/deployment/self-host\")\nprint(\"Done!\")\n", "89_typed_tool_parameters.py": "\"\"\"\nTyped Tool Parameters \u2014 list[str], dict[str, str], list[int].\n\nSince v0.22.0 (BUG-29), selectools emits proper JSON schema for typed\ncollections. OpenAI strict mode requires `items` / `additionalProperties`\nin the schema \u2014 bare `list` / `dict` without type parameters are rejected.\n\nPrerequisites: No API key needed (uses LocalProvider)\nRun: python examples/89_typed_tool_parameters.py\n\"\"\"\n\nfrom selectools import Agent\nfrom selectools.providers.stubs import LocalProvider\nfrom selectools.tools import tool\n\n\n@tool(description=\"Tag a document with labels\")\ndef tag_document(doc_id: str, tags: list[str]) -> str:\n \"\"\"Tags emits items: {type: string} in the schema.\"\"\"\n return f\"Tagged {doc_id} with {', '.join(tags)}\"\n\n\n@tool(description=\"Score items by category\")\ndef score_items(category: str, scores: list[int]) -> str:\n \"\"\"Scores emits items: {type: integer} in the schema.\"\"\"\n return f\"{category}: total={sum(scores)}, avg={sum(scores) / len(scores):.1f}\"\n\n\n@tool(description=\"Update key-value settings\")\ndef update_settings(config: dict[str, str]) -> str:\n \"\"\"Config emits additionalProperties: {type: string} in the schema.\"\"\"\n return f\"Updated {len(config)} settings: {config}\"\n\n\ndef main() -> None:\n agent = Agent(\n tools=[tag_document, score_items, update_settings],\n provider=LocalProvider(),\n )\n\n # Inspect the generated schemas\n for t in agent.tools:\n schema = t.schema()\n print(f\"\\n{t.name}:\")\n for pname, pschema in schema[\"parameters\"][\"properties\"].items():\n print(f\" {pname}: {pschema}\")\n\n # Show that list[str] produces {\"type\": \"array\", \"items\": {\"type\": \"string\"}}\n tag_schema = tag_document.schema()[\"parameters\"][\"properties\"][\"tags\"]\n assert \"items\" in tag_schema, \"list[str] must produce items in schema\"\n assert tag_schema[\"items\"][\"type\"] == \"string\"\n print(\"\\n\u2713 Typed collection schemas are correct for OpenAI strict mode\")\n\n\nif __name__ == \"__main__\":\n main()\n", "90_fallback_extended_retries.py": "\"\"\"\nFallbackProvider with Extended Retries \u2014 handle Anthropic 529, 504, Cloudflare errors.\n\nSince v0.22.0 (BUG-27), selectools recognizes these transient errors:\n- 529 Anthropic Overloaded (very common on US-West traffic)\n- 504 Gateway Timeout\n- 408 Request Timeout\n- 522/524 Cloudflare origin timeouts\n- rate_limit_exceeded (underscore form from OpenAI/Mistral)\n- overloaded/service_unavailable strings\n\nPrerequisites: OPENAI_API_KEY (or any two provider keys for real fallback)\nRun: python examples/90_fallback_extended_retries.py\n\"\"\"\n\nfrom selectools import Agent, tool\nfrom selectools.providers.fallback import FallbackProvider, _is_retriable\nfrom selectools.providers.stubs import LocalProvider\n\n\n@tool(description=\"no-op\")\ndef _noop() -> str:\n return \"ok\"\n\n\ndef main() -> None:\n # Demonstrate which errors are now retriable\n test_cases = [\n (\"429 Rate Limited\", True),\n (\"529 Anthropic Overloaded\", True),\n (\"504 Gateway Timeout\", True),\n (\"408 Request Timeout\", True),\n (\"522 Cloudflare connection timed out\", True),\n (\"524 Cloudflare origin timeout\", True),\n (\"rate_limit_exceeded: quota reached\", True),\n (\"overloaded_error: server busy\", True),\n (\"service_unavailable\", True),\n (\"400 Bad Request\", False),\n (\"401 Unauthorized\", False),\n (\"404 Not Found\", False),\n ]\n\n print(\"FallbackProvider Retriable Error Detection:\")\n print(\"-\" * 55)\n for msg, expected in test_cases:\n result = _is_retriable(Exception(msg))\n status = \"\u2713\" if result == expected else \"\u2717\"\n print(f\" {status} {msg:45s} -> {'retriable' if result else 'non-retriable'}\")\n\n # Real usage: providers=[primary, backup] with circuit breaker\n fallback = FallbackProvider(\n providers=[LocalProvider(), LocalProvider()],\n circuit_breaker_threshold=3,\n circuit_breaker_cooldown=60.0,\n on_fallback=lambda from_p, to_p, exc: print(f\" Fallback: {from_p} -> {to_p}\"),\n )\n agent = Agent(tools=[_noop], provider=fallback)\n result = agent.run(\"Hello\")\n print(f\"\\nAgent response: {result.content[:80]}\")\n\n\nif __name__ == \"__main__\":\n main()\n", "91_structured_retry_budget.py": "\"\"\"\nStructured Retry Budget \u2014 separate structured-validation retries from tool iterations.\n\nSince v0.22.0 (BUG-34), `max_iterations` controls tool-execution iterations\nand `RetryConfig.max_retries` controls structured-validation retries. They no\nlonger share a single counter.\n\nPreviously, an agent with max_iterations=3 and an LLM that failed JSON\nvalidation 3 times would terminate \u2014 even if max_retries was higher.\n\nPrerequisites: No API key needed (uses LocalProvider)\nRun: python examples/91_structured_retry_budget.py\n\"\"\"\n\nfrom pydantic import BaseModel\n\nfrom selectools import Agent, AgentConfig\nfrom selectools.agent.config_groups import RetryConfig\nfrom selectools.providers.stubs import LocalProvider\nfrom selectools.tools import tool\n\n\nclass TaskResult(BaseModel):\n status: str\n confidence: float\n\n\n@tool(description=\"A simple task\")\ndef do_task(task: str) -> str:\n return f\"Completed: {task}\"\n\n\ndef main() -> None:\n agent = Agent(\n tools=[do_task],\n provider=LocalProvider(),\n config=AgentConfig(\n max_iterations=3, # 3 tool iterations\n retry=RetryConfig(max_retries=5), # 5 structured-validation retries\n ),\n )\n\n print(\"Agent configuration:\")\n print(f\" max_iterations (tool budget): {agent.config.max_iterations}\")\n print(f\" retry.max_retries (struct budget): {agent.config.retry.max_retries}\")\n print()\n print(\"The two budgets are independent:\")\n print(\" - max_iterations=3 means the agent can call tools up to 3 times\")\n print(\" - max_retries=5 means structured output validation can fail up to 5 times\")\n print(\" - A validation failure does NOT consume a tool iteration\")\n print()\n\n # Run without response_format to show basic functionality\n result = agent.run(\"Do the task\")\n print(f\"Result: {result.content[:80]}\")\n\n\nif __name__ == \"__main__\":\n main()\n", "92_safe_parallel_pipeline.py": "\"\"\"\nSafe Parallel Pipeline \u2014 branches receive independent input copies.\n\nSince v0.22.0 (BUG-30), `parallel()` branches each receive a deep copy\nof the input. Mutations in one branch do NOT affect siblings \u2014 even under\nasyncio.gather where branches interleave at await points.\n\nPrerequisites: No API key needed\nRun: python examples/92_safe_parallel_pipeline.py\n\"\"\"\n\nimport asyncio\n\nfrom selectools.pipeline import parallel, step\n\n\n@step\ndef enrich_a(data: dict) -> dict:\n \"\"\"Branch A adds its own key.\"\"\"\n data[\"enriched_by\"] = \"branch_a\"\n data[\"a_result\"] = \"web search results\"\n return data\n\n\n@step\ndef enrich_b(data: dict) -> dict:\n \"\"\"Branch B adds its own key. Should NOT see branch A's mutation.\"\"\"\n data[\"saw_a_mutation\"] = \"enriched_by\" in data\n data[\"enriched_by\"] = \"branch_b\"\n data[\"b_result\"] = \"document search results\"\n return data\n\n\n@step\ndef merge(results: dict) -> dict:\n \"\"\"Merge results from both branches.\"\"\"\n return {\n \"from_a\": results[\"enrich_a\"][\"a_result\"],\n \"from_b\": results[\"enrich_b\"][\"b_result\"],\n \"a_saw\": results[\"enrich_a\"][\"enriched_by\"],\n \"b_saw\": results[\"enrich_b\"][\"enriched_by\"],\n }\n\n\ndef main() -> None:\n pipeline = parallel(enrich_a, enrich_b) | merge\n\n # Sync execution \u2014 pipeline.run() returns StepResult, access .output for the value\n step_result = pipeline.run({\"query\": \"quantum computing\", \"user_id\": 42})\n result = step_result.output\n print(\"Sync result:\")\n print(f\" From A: {result['from_a']}\")\n print(f\" From B: {result['from_b']}\")\n\n # The merge step received independent results from both branches.\n # If BUG-30 fix is in place, both branches worked on their own copies.\n print(\" \u2713 Both branches returned results independently\")\n\n print(\"\\n\u2713 Parallel branches are isolated \u2014 no cross-branch state corruption\")\n\n\nif __name__ == \"__main__\":\n main()\n", "93_multi_tenant_rag.py": "\"\"\"\nMulti-Tenant RAG with Permission Filters \u2014 safe metadata filtering.\n\nSince v0.22.0 (BUG-25), in-memory and BM25 stores raise NotImplementedError\nwhen you pass operator-syntax filters ({$in: [...]}) instead of silently\nreturning wrong results. Use backend stores (Chroma, Qdrant, Pinecone) for\noperator support, or use equality filters for in-memory/BM25.\n\nAlso demonstrates citation-preserving dedup (BUG-24): documents with\nidentical text but different sources are preserved as distinct citations.\n\nPrerequisites: No API key needed (uses numpy embeddings)\nRun: python examples/93_multi_tenant_rag.py\n\"\"\"\n\nfrom unittest.mock import MagicMock\n\nimport numpy as np\n\nfrom selectools.rag.bm25 import BM25\nfrom selectools.rag.stores.memory import InMemoryVectorStore\nfrom selectools.rag.vector_store import Document\n\n\ndef _mock_embedder():\n \"\"\"Create a mock embedder for demonstration.\"\"\"\n embedder = MagicMock()\n rng = np.random.RandomState(42)\n embedder.embed_query.return_value = rng.randn(8).astype(np.float32)\n embedder.embed_texts.side_effect = lambda texts: rng.randn(len(texts), 8).astype(np.float32)\n return embedder\n\n\ndef main() -> None:\n embedder = _mock_embedder()\n store = InMemoryVectorStore(embedder=embedder)\n bm25 = BM25()\n\n # Add multi-tenant documents\n docs = [\n Document(text=\"Q4 revenue was $10M\", metadata={\"tenant\": \"acme\", \"source\": \"10-K.pdf\"}),\n Document(text=\"Q4 revenue was $10M\", metadata={\"tenant\": \"globex\", \"source\": \"annual.pdf\"}),\n Document(text=\"Hiring plan for 2025\", metadata={\"tenant\": \"acme\", \"source\": \"hr.pdf\"}),\n ]\n store.add_documents(docs)\n bm25.add_documents(docs)\n\n # 1. Equality filter works everywhere\n query_emb = embedder.embed_query(\"revenue\")\n results = store.search(query_emb, top_k=10, filter={\"tenant\": \"acme\"})\n print(f\"Equality filter (tenant=acme): {len(results)} results\")\n for r in results:\n print(f\" {r.document.metadata['source']}: {r.document.text[:40]}\")\n\n # 2. Operator-syntax filters raise NotImplementedError (not silently wrong)\n print(\"\\nOperator-syntax filter on in-memory store:\")\n try:\n store.search(query_emb, filter={\"tenant\": {\"$in\": [\"acme\", \"globex\"]}})\n except NotImplementedError as e:\n print(f\" Caught: {e}\")\n\n # 3. BM25 same behavior\n print(\"\\nOperator-syntax filter on BM25:\")\n try:\n bm25.search(\"revenue\", filter={\"tenant\": {\"$in\": [\"acme\", \"globex\"]}})\n except NotImplementedError as e:\n print(f\" Caught: {e}\")\n\n # 4. Citation-preserving dedup\n results = store.search(query_emb, top_k=10, dedup=True)\n print(f\"\\nDedup search: {len(results)} results (same text, different sources preserved)\")\n for r in results:\n print(f\" {r.document.metadata.get('source', 'unknown')}: {r.document.text[:40]}\")\n\n print(\"\\n\u2713 Filters are safe \u2014 no silent permission bypass\")\n print(\"\u2713 Dedup preserves citations from different sources\")\n\n\nif __name__ == \"__main__\":\n main()\n", "94_azure_model_family.py": "\"\"\"\nAzure OpenAI with Model Family \u2014 correct token parameter for custom deployments.\n\nSince v0.22.0 (BUG-28), AzureOpenAIProvider accepts a `model_family` parameter\nso deployments with custom names (e.g., \"prod-chat\") still use the correct\n`max_completion_tokens` parameter for GPT-5-family models.\n\nPrerequisites: AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_API_KEY\nRun: python examples/94_azure_model_family.py\n\"\"\"\n\nfrom selectools import Agent\nfrom selectools.providers.azure_openai_provider import AzureOpenAIProvider\n\n\ndef main() -> None:\n # The problem: Azure deployments use custom names\n # A deployment named \"prod-chat\" running gpt-5-mini won't match\n # the \"gpt-5\" prefix that selectools uses for family detection.\n # Without model_family, this deployment would receive max_tokens\n # instead of max_completion_tokens, causing a BadRequestError.\n\n print(\"Azure OpenAI Model Family Detection:\")\n print()\n\n # Solution: pass model_family to tell selectools the underlying model\n provider = AzureOpenAIProvider.__new__(AzureOpenAIProvider)\n provider._model_family = None\n print(f\" Without model_family:\")\n print(f\" 'prod-chat' -> {provider._get_token_key('prod-chat')}\")\n print(f\" 'gpt-5-mini' -> {provider._get_token_key('gpt-5-mini')}\")\n\n provider._model_family = \"gpt-5\"\n print(f\"\\n With model_family='gpt-5':\")\n print(f\" 'prod-chat' -> {provider._get_token_key('prod-chat')}\")\n print(f\" 'gpt-5-mini' -> {provider._get_token_key('gpt-5-mini')}\")\n\n print(\"\\n\u2713 model_family overrides deployment-name-based detection\")\n print()\n print(\"Usage:\")\n print(\" provider = AzureOpenAIProvider(\")\n print(' azure_endpoint=\"https://my-resource.openai.azure.com\",')\n print(' azure_deployment=\"prod-chat\",')\n print(' model_family=\"gpt-5\", # Underlying model family')\n print(\" )\")\n\n\nif __name__ == \"__main__\":\n main()\n", "95_loop_detection.py": "\"\"\"\nLoop Detection \u2014 catch pathological tool-call patterns before max_iterations.\n\nSince v0.22.0, `AgentConfig.loop_detector` enables three composable detectors:\n\n - RepeatDetector: same (tool, args) N times in a row\n - StallDetector: same (tool, result) N times in a row \u2014 polling with no progress\n - PingPongDetector: a cycle of length K repeats M times\n\nOn detection, the agent either raises LoopDetectedError (default) or injects\na corrective system message and continues (LoopPolicy.INJECT_MESSAGE).\n\nPrerequisites: No API key needed (uses a scripted provider)\nRun: python examples/95_loop_detection.py\n\"\"\"\n\nfrom dataclasses import dataclass\nfrom typing import Any, List, Optional, Tuple\n\nfrom selectools import (\n Agent,\n AgentConfig,\n LoopDetectedError,\n LoopDetector,\n LoopPolicy,\n RepeatDetector,\n)\nfrom selectools.tools import tool\nfrom selectools.types import Message, Role, ToolCall\nfrom selectools.usage import UsageStats\n\n\n@tool(description=\"Search the web\")\ndef search(query: str) -> str:\n return \"no results found\"\n\n\n@dataclass\nclass _RepeatingProvider:\n \"\"\"Minimal provider that always returns the same tool call \u2014 simulates a stuck LLM.\"\"\"\n\n name: str = \"repeating\"\n supports_streaming: bool = False\n supports_async: bool = True\n\n def _next(self) -> Tuple[Message, UsageStats]:\n call = ToolCall(tool_name=\"search\", parameters={\"query\": \"cats\"}, id=\"tc_1\")\n msg = Message(role=Role.ASSISTANT, content=\"\", tool_calls=[call])\n return msg, UsageStats(model=\"fake\", provider=self.name)\n\n def complete(\n self,\n *,\n model: str = \"fake\",\n system_prompt: str = \"\",\n messages: List[Message],\n tools: Optional[List[Any]] = None,\n temperature: float = 0.0,\n max_tokens: int = 1000,\n timeout: Optional[float] = None,\n ) -> Tuple[Message, UsageStats]:\n return self._next()\n\n async def acomplete(self, **kwargs: Any) -> Tuple[Message, UsageStats]:\n return self._next()\n\n\ndef demo_raise() -> None:\n \"\"\"Default policy: raise LoopDetectedError.\"\"\"\n agent = Agent(\n tools=[search],\n provider=_RepeatingProvider(),\n config=AgentConfig(\n max_iterations=20,\n loop_detector=LoopDetector.default(),\n ),\n )\n try:\n agent.run(\"Find something\")\n except LoopDetectedError as exc:\n print(f\"[RAISE] detector={exc.detector} details={exc.details}\")\n\n\ndef demo_inject_message() -> None:\n \"\"\"INJECT_MESSAGE policy: add a corrective system message, keep looping until max_iterations.\"\"\"\n detector = LoopDetector(\n detectors=[RepeatDetector(threshold=3)],\n policy=LoopPolicy.INJECT_MESSAGE,\n inject_message=\"You are repeating yourself. Try a different approach.\",\n )\n agent = Agent(\n tools=[search],\n provider=_RepeatingProvider(),\n config=AgentConfig(max_iterations=5, loop_detector=detector),\n )\n result = agent.run(\"Find something\")\n print(f\"[INJECT] iterations={result.iterations} no exception raised\")\n\n\ndef main() -> None:\n print(\"Loop detection \u2014 three patterns, two policies.\\n\")\n demo_raise()\n demo_inject_message()\n\n\nif __name__ == \"__main__\":\n main()\n", "96_supabase_session_store.py": "#!/usr/bin/env python3\n\"\"\"\nSupabase Session Store \u2014 Postgres-backed sessions via Supabase PostgREST.\n\nDemonstrates SupabaseSessionStore: ConversationMemory persisted to a Supabase\nPostgres table as JSONB, with idempotent upserts and optional namespace\nisolation. Fourth backend alongside JSON file, SQLite, and Redis.\n\nNo API key needed. Runs entirely offline with the built-in LocalProvider and\nan in-process fake Supabase client so the demo works without a live project.\nSwap the fake for `supabase.create_client(SUPABASE_URL, SERVICE_ROLE_KEY)` in\nproduction.\n\nTable DDL (run once in your Supabase project):\n\n create table if not exists public.selectools_sessions (\n session_id text primary key,\n memory_json jsonb not null,\n message_count integer not null default 0,\n created_at timestamptz not null default now(),\n updated_at timestamptz not null default now()\n );\n alter table public.selectools_sessions enable row level security;\n\nPrerequisites: pip install selectools[supabase]\nRun: python examples/96_supabase_session_store.py\n\"\"\"\n\nfrom __future__ import annotations\n\nimport sys\nfrom datetime import datetime, timezone\nfrom typing import Any, Dict, List, Optional\nfrom unittest.mock import MagicMock, patch\n\nfrom selectools import Agent, AgentConfig, ConversationMemory, Message, Role, tool\nfrom selectools.providers.stubs import LocalProvider\nfrom selectools.sessions import SupabaseSessionStore\n\n# \u2500\u2500 Minimal in-process Supabase fake (demo only) \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n# In production, replace this with:\n# from supabase import create_client\n# client = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)\n\n\nclass _FakeResponse:\n def __init__(self, data: Any) -> None:\n self.data = data\n\n\nclass _FakeQuery:\n def __init__(self, store: Dict[str, Dict[str, Any]], table: str) -> None:\n self._store, self._table, self._op = store, table, \"select\"\n self._payload: Optional[Dict[str, Any]] = None\n self._filters: List[tuple] = []\n self._cols = \"*\"\n self._limit: Optional[int] = None\n\n def select(self, cols: str = \"*\") -> \"_FakeQuery\":\n self._cols = cols\n return self\n\n def eq(self, col: str, val: Any) -> \"_FakeQuery\":\n self._filters.append((col, val))\n return self\n\n def limit(self, n: int) -> \"_FakeQuery\":\n self._limit = n\n return self\n\n def upsert(self, payload: Dict[str, Any], **_: Any) -> \"_FakeQuery\":\n self._op, self._payload = \"upsert\", payload\n return self\n\n def delete(self) -> \"_FakeQuery\":\n self._op = \"delete\"\n return self\n\n def execute(self) -> _FakeResponse:\n rows = self._store.setdefault(self._table, {})\n if self._op == \"upsert\":\n assert self._payload is not None\n key = self._payload[\"session_id\"]\n existing = rows.get(key, {})\n merged = {**existing, **self._payload}\n if \"created_at\" not in merged:\n merged[\"created_at\"] = datetime.now(timezone.utc).isoformat()\n rows[key] = merged\n return _FakeResponse([merged])\n\n matching = [r for r in rows.values() if all(r.get(c) == v for c, v in self._filters)]\n if self._op == \"delete\":\n for r in matching:\n rows.pop(r[\"session_id\"], None)\n return _FakeResponse(matching)\n\n if self._limit is not None:\n matching = matching[: self._limit]\n if self._cols != \"*\":\n cols = [c.strip() for c in self._cols.split(\",\")]\n matching = [{c: r[c] for c in cols if c in r} for r in matching]\n return _FakeResponse(matching)\n\n\nclass _FakeSupabaseClient:\n def __init__(self) -> None:\n self._data: Dict[str, Dict[str, Any]] = {}\n\n def table(self, name: str) -> _FakeQuery:\n return _FakeQuery(self._data, name)\n\n\n@tool(description=\"Get the current weather for a city\")\ndef get_weather(city: str) -> str:\n return {\"paris\": \"18C, sunny\", \"london\": \"12C, cloudy\"}.get(city.lower(), f\"No data for {city}\")\n\n\ndef main() -> None:\n client = _FakeSupabaseClient()\n # SupabaseSessionStore lazily imports `supabase`. Patch the import so this\n # demo runs without the package installed. In real use you just do:\n # store = SupabaseSessionStore(client=client)\n with patch.dict(sys.modules, {\"supabase\": MagicMock()}):\n store = SupabaseSessionStore(client=client, table_name=\"selectools_sessions\")\n\n session_id = \"demo-session\"\n\n # \u2500\u2500 1. Agent auto-save/load via AgentConfig \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(\"=== Turn 1 \u2014 agent auto-saves to Supabase ===\\n\")\n memory = ConversationMemory(max_messages=20)\n agent = Agent(\n tools=[get_weather],\n provider=LocalProvider(),\n config=AgentConfig(max_iterations=2, session_store=store, session_id=session_id),\n memory=memory,\n )\n result = agent.run([Message(role=Role.USER, content=\"Weather in Paris?\")])\n print(f\"Agent: {result.content}\")\n print(f\"Row upserted: {store.exists(session_id)}\\n\")\n\n print(\"=== Turn 2 \u2014 fresh agent loads from the same session_id ===\\n\")\n restored = store.load(session_id)\n assert restored is not None\n print(f\"Restored {len(restored)} messages from Postgres\")\n agent2 = Agent(\n tools=[get_weather],\n provider=LocalProvider(),\n config=AgentConfig(max_iterations=2, session_store=store, session_id=session_id),\n memory=restored,\n )\n agent2.run([Message(role=Role.USER, content=\"Now London.\")])\n print(f\"After turn 2: {len(restored)} messages persisted\\n\")\n\n # \u2500\u2500 2. Namespace isolation (store-level, not agent-level) \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n # Use namespaces to keep multiple agents' sessions from colliding on the\n # same session_id. Pass namespace= directly to save/load/exists/delete.\n print(\"=== Namespace isolation ===\\n\")\n mem_a = ConversationMemory(max_messages=10)\n mem_a.add(Message(role=Role.USER, content=\"agent-a note\"))\n mem_b = ConversationMemory(max_messages=10)\n mem_b.add(Message(role=Role.USER, content=\"agent-b note\"))\n store.save(\"shared\", mem_a, namespace=\"agent-a\")\n store.save(\"shared\", mem_b, namespace=\"agent-b\")\n loaded_a = store.load(\"shared\", namespace=\"agent-a\")\n loaded_b = store.load(\"shared\", namespace=\"agent-b\")\n assert loaded_a is not None and loaded_b is not None\n print(f\" agent-a sees: {loaded_a.get_history()[0].content}\")\n print(f\" agent-b sees: {loaded_b.get_history()[0].content}\")\n print(f\" Exists bare 'shared': {store.exists('shared')} (no collision)\\n\")\n\n # \u2500\u2500 3. Branch a session \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(\"=== Branching ===\\n\")\n store.branch(session_id, f\"{session_id}-fork\")\n for meta in store.list():\n print(f\" id={meta.session_id} messages={meta.message_count}\")\n\n # \u2500\u2500 4. Cleanup \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(\"\\n=== Cleanup ===\")\n store.delete(session_id)\n store.delete(f\"{session_id}-fork\")\n store.delete(\"shared\", namespace=\"agent-a\")\n store.delete(\"shared\", namespace=\"agent-b\")\n print(f\"Rows remaining: {len(store.list())}\")\n\n\nif __name__ == \"__main__\":\n main()\n", "97_agent_as_api.py": "#!/usr/bin/env python3\n\"\"\"\nAgent-as-API \u2014 auto-generated production REST endpoints for any Agent.\n\nDemonstrates AgentAPI: one line turns an Agent (or several) into a\nproduction-ready Starlette ASGI app with standardized JSON schemas,\nsession persistence via any SessionStore backend, per-user isolation\nthrough the user_id header, and optional bearer-token auth.\n\nEndpoints generated:\n\n POST /v1/chat \u2014 single-turn completion (JSON)\n POST /v1/chat/stream \u2014 streaming completion (SSE)\n POST /v1/sessions \u2014 create session\n GET /v1/sessions/{id} \u2014 get session history\n DELETE /v1/sessions/{id} \u2014 delete session\n GET /v1/health \u2014 health check (never requires auth)\n\nNo API key needed. Runs entirely offline with the built-in LocalProvider\nand Starlette's TestClient. In production, deploy with uvicorn:\n\n # api.py\n from selectools.serve import AgentAPI\n app = AgentAPI(agents=[my_agent], auth_key=\"sk-...\")\n # uvicorn api:app --port 8000\n\nOr straight from a YAML config:\n\n selectools serve agent.yaml --api --port 8000\n\nPrerequisites: pip install selectools[serve]\nRun: python examples/97_agent_as_api.py\n\"\"\"\n\nfrom __future__ import annotations\n\nimport json\n\nfrom selectools import Agent, AgentConfig, tool\nfrom selectools.providers.stubs import LocalProvider\nfrom selectools.serve import AgentAPI\n\nAUTH_KEY = \"sk-demo-key\"\n\n\n@tool()\ndef word_count(text: str) -> str:\n \"\"\"Count the words in a text.\"\"\"\n return f\"{len(text.split())} words\"\n\n\ndef build_app() -> AgentAPI:\n support = Agent(\n tools=[word_count],\n provider=LocalProvider(),\n config=AgentConfig(name=\"support\", model=\"local-model\"),\n )\n billing = Agent(\n tools=[word_count],\n provider=LocalProvider(),\n config=AgentConfig(name=\"billing\", model=\"local-model\"),\n )\n # Multi-agent: route requests with the optional \"agent\" field.\n # session_store accepts any SessionStore (JSON file, SQLite, Redis,\n # Supabase); the default is an in-memory store.\n return AgentAPI(agents=[support, billing], auth_key=AUTH_KEY)\n\n\ndef main() -> None:\n from starlette.testclient import TestClient\n\n app = build_app()\n client = TestClient(app)\n auth = {\"Authorization\": f\"Bearer {AUTH_KEY}\"}\n\n print(\"=== GET /v1/health (no auth required) ===\")\n print(json.dumps(client.get(\"/v1/health\").json(), indent=2))\n\n print(\"\\n=== POST /v1/chat without auth -> 401 ===\")\n r = client.post(\"/v1/chat\", json={\"input\": \"hello\"})\n print(r.status_code, json.dumps(r.json()))\n\n print(\"\\n=== POST /v1/chat (default agent, new session) ===\")\n r = client.post(\n \"/v1/chat\",\n json={\"input\": \"How many words is 'the quick brown fox'?\"},\n headers={**auth, \"user_id\": \"alice\"},\n )\n body = r.json()\n session_id = body[\"session_id\"]\n print(json.dumps(body, indent=2))\n\n print(\"\\n=== POST /v1/chat (continue session, route to 'billing') ===\")\n r = client.post(\n \"/v1/chat\",\n json={\"input\": \"And my invoice?\", \"session_id\": session_id, \"agent\": \"billing\"},\n headers={**auth, \"user_id\": \"alice\"},\n )\n print(json.dumps(r.json(), indent=2))\n\n print(\"\\n=== GET /v1/sessions/{id} (alice sees her history) ===\")\n r = client.get(f\"/v1/sessions/{session_id}\", headers={**auth, \"user_id\": \"alice\"})\n print(json.dumps(r.json(), indent=2))\n\n print(\"\\n=== GET /v1/sessions/{id} as bob -> 404 (per-user isolation) ===\")\n r = client.get(f\"/v1/sessions/{session_id}\", headers={**auth, \"user_id\": \"bob\"})\n print(r.status_code, json.dumps(r.json()))\n\n print(\"\\n=== POST /v1/chat/stream (SSE) ===\")\n r = client.post(\n \"/v1/chat/stream\",\n json={\"input\": \"stream me\"},\n headers={**auth, \"user_id\": \"alice\"},\n )\n for line in r.text.splitlines():\n if line.startswith(\"data: \"):\n print(line)\n\n print(\"\\n=== DELETE /v1/sessions/{id} ===\")\n r = client.delete(f\"/v1/sessions/{session_id}\", headers={**auth, \"user_id\": \"alice\"})\n print(r.status_code, json.dumps(r.json()))\n\n\nif __name__ == \"__main__\":\n main()\n", "98_knowledge_backend.py": "#!/usr/bin/env python3\n\"\"\"\nKnowledge Backends \u2014 persist KnowledgeMemory across ephemeral deploys.\n\nDemonstrates the KnowledgeBackend protocol: KnowledgeMemory keeps using a\nfast local directory as scratch space, while a backend (Supabase or Redis)\npersists a snapshot of that directory between requests. Railway, Lambda, and\nCloud Run wipe /tmp between deploys \u2014 the backend doesn't.\n\nNo API key needed. Runs entirely offline with an in-process fake Supabase\nclient so the demo works without a live project. Swap the fake for\n`supabase.create_client(SUPABASE_URL, SERVICE_ROLE_KEY)` in production.\n\nTable DDL (run once in your Supabase project):\n\n create table if not exists public.selectools_knowledge (\n key text primary key,\n data text not null,\n updated_at timestamptz not null default now()\n );\n alter table public.selectools_knowledge enable row level security;\n\nPrerequisites: pip install selectools[supabase]\nRun: python examples/98_knowledge_backend.py\n\"\"\"\n\nfrom __future__ import annotations\n\nimport tempfile\nfrom typing import Any, Dict, List, Optional\nfrom unittest.mock import MagicMock, patch\n\nfrom selectools import KnowledgeMemory\nfrom selectools.knowledge_backends import SupabaseKnowledgeBackend\n\n# \u2500\u2500 Minimal in-process Supabase fake (demo only) \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n\n\nclass _FakeResponse:\n def __init__(self, data: Any) -> None:\n self.data = data\n\n\nclass _FakeQuery:\n def __init__(self, rows: Dict[str, Dict[str, Any]]) -> None:\n self._rows = rows\n self._filters: List[tuple] = []\n self._payload: Optional[Dict[str, Any]] = None\n self._conflict: Optional[str] = None\n\n def select(self, _cols: str = \"*\") -> \"_FakeQuery\":\n return self\n\n def eq(self, col: str, val: Any) -> \"_FakeQuery\":\n self._filters.append((col, val))\n return self\n\n def limit(self, _n: int) -> \"_FakeQuery\":\n return self\n\n def upsert(self, payload: Dict[str, Any], on_conflict: str = \"key\") -> \"_FakeQuery\":\n self._payload = payload\n self._conflict = on_conflict\n return self\n\n def execute(self) -> _FakeResponse:\n if self._payload is not None:\n self._rows[self._payload[self._conflict]] = dict(self._payload)\n return _FakeResponse([self._payload])\n rows = list(self._rows.values())\n for col, val in self._filters:\n rows = [r for r in rows if r.get(col) == val]\n return _FakeResponse(rows)\n\n\nclass _FakeSupabaseClient:\n def __init__(self) -> None:\n self._tables: Dict[str, Dict[str, Dict[str, Any]]] = {}\n\n def table(self, name: str) -> _FakeQuery:\n return _FakeQuery(self._tables.setdefault(name, {}))\n\n\ndef main() -> None:\n client = _FakeSupabaseClient()\n\n # The fake client means no real `supabase` package is needed for the demo.\n with patch.dict(\"sys.modules\", {\"supabase\": MagicMock()}):\n # \u2500\u2500 Deploy 1: remember facts, backend persists automatically \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n with tempfile.TemporaryDirectory() as scratch1:\n memory = KnowledgeMemory(\n directory=scratch1,\n backend=SupabaseKnowledgeBackend(client, key=\"user-123\"),\n )\n memory.remember(\"User prefers dark mode\", persistent=True, importance=0.9)\n memory.remember(\"Working on the Q3 report\", category=\"context\", importance=0.6)\n print(\"Deploy 1 context:\")\n print(memory.build_context())\n\n # \u2500\u2500 Deploy 2: fresh /tmp, same backend key \u2014 nothing was lost \u2500\u2500\u2500\u2500\u2500\u2500\n with tempfile.TemporaryDirectory() as scratch2:\n memory = KnowledgeMemory(\n directory=scratch2,\n backend=SupabaseKnowledgeBackend(client, key=\"user-123\"),\n )\n print(\"\\nDeploy 2 context (restored from Supabase):\")\n print(memory.build_context())\n assert \"dark mode\" in memory.build_context()\n\n # \u2500\u2500 Keys isolate users \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n with tempfile.TemporaryDirectory() as scratch3:\n other = KnowledgeMemory(\n directory=scratch3,\n backend=SupabaseKnowledgeBackend(client, key=\"user-456\"),\n )\n print(\"\\nDifferent key starts empty:\", repr(other.build_context()))\n\n print(\"\\nDone. In production, create the client with supabase.create_client(...).\")\n\n\nif __name__ == \"__main__\":\n main()\n", "99_tool_results_artifacts.py": "#!/usr/bin/env python3\n\"\"\"\nTyped ToolResult returns + artifact side-channel (issue #59).\n\nTwo conventions every tool-using agent otherwise re-invents:\n\n1. ToolResult \u2014 frozen dataclass base for typed tool returns. Subclasses set\n a ``kind`` discriminator as a ClassVar; the serializer re-injects it into\n the JSON the model sees (ClassVar fields are dropped by asdict()).\n Built-ins: Ambiguous, NotFound. Note the epistemics: ``not_found`` means\n \"this tool observed no match from this source at this time\", not \"the\n entity does not exist\".\n\n2. emit_artifact() \u2014 tools that produce files (charts, PDFs, exports) attach\n them out-of-band instead of stuffing URLs into the reply string. The agent\n drains the per-run collector into AgentResult.artifacts, where channel\n layers (chat, email, Slack) deliver them. The shape is richer than a URL\n on purpose: URLs rot and signed links expire \u2014 sha256 + size let consumers\n identify an artifact without storing its body.\n\nNo API key needed. Runs offline with a scripted provider.\n\nRun: python examples/99_tool_results_artifacts.py\n\"\"\"\n\nfrom __future__ import annotations\n\nimport hashlib\nfrom dataclasses import dataclass\nfrom typing import Any, ClassVar, List, Tuple\n\nfrom selectools import Agent, AgentConfig, Message, NotFound, Role, ToolCall, emit_artifact, tool\nfrom selectools.providers.base import Provider\nfrom selectools.results import ToolResult\nfrom selectools.usage import UsageStats\n\n# ---------------------------------------------------------------------------\n# Tools\n# ---------------------------------------------------------------------------\n\n_CUSTOMERS = {\"acme\": {\"id\": 1, \"name\": \"Acme Corp\", \"plan\": \"enterprise\"}}\n\n\n@dataclass(frozen=True)\nclass CustomerFound(ToolResult):\n \"\"\"A user-defined typed result \u2014 kind survives serialization.\"\"\"\n\n kind: ClassVar[str] = \"customer_found\"\n\n customer_id: int\n name: str\n plan: str\n\n\n@tool()\ndef find_customer(query: str) -> ToolResult:\n \"\"\"Look up a customer by name.\"\"\"\n row = _CUSTOMERS.get(query.lower())\n if row is None:\n # Observation, not a truth claim: no match from THIS source NOW.\n return NotFound(entity=\"customer\", query=query)\n return CustomerFound(customer_id=row[\"id\"], name=row[\"name\"], plan=row[\"plan\"])\n\n\n@tool()\ndef render_chart(title: str) -> str:\n \"\"\"Render a revenue chart as PNG.\"\"\"\n body = f\"\".encode()\n emit_artifact(\n f\"https://files.example.com/charts/{title}.png\",\n mime_type=\"image/png\",\n filename=f\"{title}.png\",\n sha256=hashlib.sha256(body).hexdigest(),\n size=len(body),\n role=\"primary\",\n retention=\"30d\",\n )\n # The reply string stays LLM-friendly; the file travels on the side.\n return f\"Rendered chart '{title}'.\"\n\n\n# ---------------------------------------------------------------------------\n# Scripted offline provider\n# ---------------------------------------------------------------------------\n\n\nclass ScriptedProvider(Provider):\n \"\"\"Issues a fixed sequence of tool calls, then a final answer.\"\"\"\n\n name = \"scripted\"\n supports_streaming = False\n supports_async = False\n\n def __init__(self, tool_calls: List[ToolCall]) -> None:\n self.default_model = \"scripted\"\n self._tool_calls = tool_calls\n self._turn = 0\n\n def complete(self, **kwargs: Any) -> Tuple[Message, UsageStats]:\n usage = UsageStats(0, 0, 0, 0.0, model=\"scripted\", provider=\"scripted\")\n self._turn += 1\n if self._turn == 1:\n return Message(role=Role.ASSISTANT, content=\"\", tool_calls=self._tool_calls), usage\n return Message(role=Role.ASSISTANT, content=\"Here is the Q2 report.\"), usage\n\n\n# ---------------------------------------------------------------------------\n# Demo\n# ---------------------------------------------------------------------------\n\n\ndef main() -> None:\n # 1. Typed results: kind survives into the JSON the LLM sees.\n print(\"== ToolResult serialization ==\")\n print(\"hit: \", find_customer.execute({\"query\": \"acme\"}))\n print(\"miss:\", find_customer.execute({\"query\": \"globex\"}))\n\n # 2. Artifact side-channel through a full agent run.\n print(\"\\n== Artifact side-channel ==\")\n provider = ScriptedProvider(\n [ToolCall(tool_name=\"render_chart\", parameters={\"title\": \"q2-revenue\"}, id=\"c1\")]\n )\n agent = Agent(\n tools=[find_customer, render_chart],\n provider=provider,\n config=AgentConfig(max_iterations=3),\n )\n result = agent.run(\"Chart Q2 revenue for Acme\")\n print(\"reply: \", result.content)\n for artifact in result.artifacts:\n print(\"artifact: \", artifact.url)\n print(\" mime:\", artifact.mime_type, \"| size:\", artifact.size, \"bytes\")\n print(\" sha256:\", (artifact.sha256 or \"\")[:16], \"... | role:\", artifact.role)\n\n\nif __name__ == \"__main__\":\n main()\n"}; let ac='all'; function flt(){const q=document.getElementById('si').value.toLowerCase();let c=0;document.querySelectorAll('.ec').forEach(d=>{const t=d.dataset.title,f=d.dataset.file,cats=d.dataset.cats;const cm=ac==='all'||cats.includes(ac);const sm=!q||t.includes(q)||f.includes(q)||cats.includes(q);const s=cm&&sm;d.style.display=s?'':'none';if(s)c++});document.getElementById('rc').textContent='# '+c+' files match'} document.querySelectorAll('.ex-rail__seg').forEach(b=>{b.addEventListener('click',()=>{document.querySelectorAll('.ex-rail__seg').forEach(x=>{x.classList.remove('on');x.setAttribute('aria-selected','false')});b.classList.add('on');b.setAttribute('aria-selected','true');ac=b.dataset.cat;b.style.animation='none';requestAnimationFrame(()=>{b.style.animation='exec-stamp 0.6s var(--exec-ease-soft)'});flt();syncPrompt()});}); (function(){const r=document.getElementById('ex-rail');if(!r)return;const io=new IntersectionObserver((ents)=>{ents.forEach(e=>{if(e.isIntersecting){r.classList.add('in-view');io.disconnect()}})},{rootMargin:'0px 0px -20% 0px'});io.observe(r)})(); function hl(s){s=s.replace(/&/g,'&').replace(//g,'>');s=s.replace(/\b(from|import|def|class|return|if|elif|else|for|while|with|as|try|except|finally|raise|yield|async|await|and|or|not|in|is|True|False|None|lambda|pass|break|continue)\b/g,'$1');s=s.replace(/(#[^\n]*)/g,'$1');s=s.replace(/(@\w+(?:\([^)]*\))?)/g,'$1');return s} function toggle(h){const c=h.closest('.ec'),b=c.querySelector('.eb'),p=c.querySelector('.ep');c.classList.toggle('op');const open=c.classList.contains('op');b.style.display=open?'':'none';h.setAttribute('aria-expanded',open?'true':'false');if(open&&!p.dataset.loaded){p.innerHTML=hl(SRC[c.dataset.file]||'');p.dataset.loaded='1'}} function cpSrc(b){const f=b.closest('.ec').dataset.file;navigator.clipboard.writeText(SRC[f]||'');b.textContent='Copied!';setTimeout(()=>b.textContent='Copy',1500)} function syncPrompt(){const q=document.getElementById('si').value;document.getElementById('ex-grep').textContent=q?' | grep -i '+q:'';document.getElementById('ex-flags').textContent=ac==='all'?'':' --tags '+ac} function typeLine(target,text,perChar,done){let i=0;const tick=()=>{if(i<=text.length){target.textContent=text.slice(0,i);i++;setTimeout(tick,perChar)}else if(done){done()}};tick()} (function bootPrompt(){const cmd=document.getElementById('ex-cmd');if(!cmd)return;const reduced=window.matchMedia('(prefers-reduced-motion: reduce)').matches;if(reduced){cmd.textContent='ls examples/';syncPrompt();return}typeLine(cmd,'ls examples/',35,syncPrompt)})(); document.addEventListener('keydown',(e)=>{if(e.key!=='/')return;const t=e.target;if(t&&(t.tagName==='INPUT'||t.tagName==='TEXTAREA'||t.isContentEditable))return;e.preventDefault();const si=document.getElementById('si');if(si)si.focus()}); document.querySelectorAll('.ex-row').forEach(r=>{r.addEventListener('keydown',(e)=>{if(e.key==='Enter'||e.key===' '){e.preventDefault();toggle(r)}})});