synth-agent-sdk 0.8.0
Autonomous agents, engineered. A Python SDK for building production-grade AI agents and multi-agent systems.
Requires Python: >=3.10

# Synth

> Autonomous agents, engineered.

**Latest Version:** 0.6.1 | [PyPI](https://pypi.org/project/synth-agent-sdk/) | [Changelog](CHANGELOG.md)

A Python SDK for building production-grade AI agents and multi-agent systems. From a 3-line single agent to complex, stateful, resumable multi-agent graphs — with model-agnostic provider support, streaming, observability, evaluation, and guardrails out of the box.

---

## Table of Contents

1. [What is Synth?](#what-is-synth)
2. [Installation](#installation)
3. [Quick Start](#quick-start)
4. [Core Concepts](#core-concepts)
5. [Creating an Agent](#creating-an-agent)
6. [Tools](#tools)
7. [Running Your Agent](#running-your-agent)
8. [Streaming](#streaming)
9. [Model Providers](#model-providers)
10. [Memory](#memory)
11. [Guards](#guards)
12. [Structured Output](#structured-output)
13. [Pipelines](#pipelines)
14. [Graphs](#graphs)
15. [Human-in-the-Loop](#human-in-the-loop)
16. [Agent Teams](#agent-teams)
17. [Tracing](#tracing)
18. [Checkpointing](#checkpointing)
19. [Evaluation](#evaluation)
20. [CLI Commands](#cli-commands)
21. [Deploying to AWS AgentCore](#deploying-to-aws-agentcore)
22. [Error Handling](#error-handling)
23. [Environment Variables](#environment-variables)
24. [FAQ](#faq)

---

## What is Synth?

Synth is a Python library for building AI-powered agents. An agent uses a large language model (Claude, GPT, Gemini, etc.) to understand instructions, make decisions, and take actions — calling functions, searching databases, generating reports, or coordinating with other agents.

Synth handles the plumbing (provider communication, conversation management, retries, cost tracking) so you focus on what your agent actually does.

---

## Installation

Requires Python 3.10+.

```bash
pip install synth-agent-sdk[anthropic]     # Anthropic Claude (recommended)
```

Other options:

```bash
pip install synth-agent-sdk[quickstart]    # Claude + GPT (tutorials/demos)
pip install synth-agent-sdk[openai]        # OpenAI GPT
pip install synth-agent-sdk[google]        # Google Gemini
pip install synth-agent-sdk[ollama]        # Local Ollama models
pip install synth-agent-sdk[bedrock]       # AWS Bedrock
pip install synth-agent-sdk[agentcore]     # AWS AgentCore deployment
pip install synth-agent-sdk[all]           # All providers
```

Set your API key:

```bash
export ANTHROPIC_API_KEY="your-key-here"   # Claude
export OPENAI_API_KEY="your-key-here"      # GPT
export GOOGLE_API_KEY="your-key-here"      # Gemini
# AWS Bedrock uses standard IAM credentials — no Synth-specific key needed
```

Verify your setup:

```bash
synth doctor
```

---

## Quick Start

```python
from synth import Agent

agent = Agent(model="claude-sonnet-4-5", instructions="You are a helpful assistant.")
result = agent.run("What is the capital of France?")
print(result.text)
# => "The capital of France is Paris."
```

---

## Core Concepts

| Concept | What It Is |
|---------|-----------|
| `Agent` | The main building block. Wraps an AI model with tools, memory, and guards. |
| `Tool` | A Python function your agent can call. |
| `ToolKit` | A bundle of related tools. |
| `RunResult` | Returned by `agent.run()` — text, token usage, cost, latency, trace. |
| `Memory` | Lets your agent remember previous conversations. |
| `Guard` | A safety rule applied to input or output. |
| `Pipeline` | Chains agents sequentially. |
| `Graph` | A workflow with branching, loops, and conditional logic. |
| `AgentTeam` | Multiple agents coordinated by an orchestrator. |
| `Trace` | A detailed record of everything that happened during a run. |
| `Checkpoint` | A saved snapshot of a run's state for resumption. |

---

## Creating an Agent

```python
from synth import Agent, Guard, Memory

agent = Agent(
    model="claude-sonnet-4-5",        # AI model to use
    instructions="You are helpful.",   # System prompt
    tools=[my_tool, my_toolkit],      # Optional tools
    memory=Memory.thread(),           # Optional memory
    guards=[Guard.no_pii_output()],   # Optional safety rules
    output_schema=MyModel,            # Optional Pydantic schema
    max_retries=3,                    # Retry on transient errors
    retry_backoff=1.0,                # Base delay between retries (seconds)
)
```

All parameters except `model` are optional. Default model is `claude-sonnet-4-5`.

---

## Tools

Tools are Python functions your agent can call. Mark them with `@tool` — Synth auto-generates JSON schemas from type hints and docstrings.

```python
from synth import tool

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"The weather in {city} is sunny, 72°F."

agent = Agent(
    model="claude-sonnet-4-5",
    instructions="You are a weather assistant.",
    tools=[get_weather],
)
```

Rules: every parameter needs a type annotation, and the function needs a docstring. Missing either raises `ToolDefinitionError` immediately.

Group related tools with `ToolKit`:

```python
from synth import ToolKit

math_tools = ToolKit([add, multiply, divide])
agent = Agent(model="gpt-4o", tools=[math_tools, get_weather])
```

Inspect tool calls after a run:

```python
for tc in result.tool_calls:
    print(f"{tc.name}({tc.args}) → {tc.result}  [{tc.latency_ms:.1f}ms]")
```

---

## Running Your Agent

**Synchronous:**

```python
result = agent.run("Explain quantum computing in simple terms.")
print(result.text)        # Response text
print(result.tokens)      # TokenUsage(input, output, total)
print(result.cost)        # Estimated cost in USD
print(result.latency_ms)  # Latency in milliseconds
print(result.tool_calls)  # Tools that were called
print(result.trace)       # Full execution trace
print(result.output)      # Parsed structured output (if output_schema set)
```

**Asynchronous:**

```python
import asyncio

async def main():
    result = await agent.arun("What is 2 + 2?")
    print(result.text)

asyncio.run(main())
```

---

## Streaming

```python
from synth import TokenEvent, ToolCallEvent, ToolResultEvent, DoneEvent, ErrorEvent

for event in agent.stream("Write a short poem about coding."):
    if isinstance(event, TokenEvent):
        print(event.text, end="", flush=True)
    elif isinstance(event, ToolCallEvent):
        print(f"\n[Calling: {event.name}]")
    elif isinstance(event, DoneEvent):
        print(f"\n\nTokens: {event.result.tokens.total_tokens}")
```

Async streaming:

```python
async for event in agent.astream("Write a haiku."):
    if isinstance(event, TokenEvent):
        print(event.text, end="", flush=True)
```

| Event | When |
|-------|------|
| `TokenEvent` | Model produced a text token |
| `ToolCallEvent` | Model decided to call a tool |
| `ToolResultEvent` | Tool finished executing |
| `ThinkingEvent` | Model produced a reasoning token |
| `DoneEvent` | Stream completed — contains full `RunResult` |
| `ErrorEvent` | Something went wrong |

---

## Model Providers

Switch providers by changing the `model` string — no other code changes needed.

| Provider | Model String Examples | Extra | API Key |
|----------|----------------------|-------|---------|
| Anthropic | `"claude-sonnet-4-5"`, `"claude-haiku-3-5"` | `synth[anthropic]` | `ANTHROPIC_API_KEY` |
| OpenAI | `"gpt-4o"`, `"gpt-4o-mini"` | `synth[openai]` | `OPENAI_API_KEY` |
| Google | `"gemini-2.0-flash"` | `synth[google]` | `GOOGLE_API_KEY` |
| Ollama | `"ollama/llama3"`, `"ollama/mistral"` | `synth[ollama]` | None (local) |
| AWS Bedrock | `"bedrock/claude-sonnet-4-5"` | `synth[bedrock]` | AWS IAM |

Custom endpoint:

```python
agent = Agent(model="my-model", base_url="https://my-proxy.example.com/v1")
```

---

## Memory

By default each `run()` is stateless. Add memory to persist conversations.

**Thread memory** (in-process, fast):

```python
agent = Agent(model="claude-sonnet-4-5", memory=Memory.thread())

agent.run("My name is Alice.", thread_id="user-123")
result = agent.run("What's my name?", thread_id="user-123")
print(result.text)  # "Your name is Alice."
```

**Persistent memory** (Redis, survives restarts):

```python
agent = Agent(model="gpt-4o", memory=Memory.persistent("redis://localhost:6379"))
```

**Semantic memory** (vector embeddings, retrieves most relevant context):

```python
agent = Agent(model="gemini-2.0-flash", memory=Memory.semantic(embedder=my_embedder_fn))
```

---

## Guards

Declarative safety rules applied automatically to every run.

```python
from synth import Guard

agent = Agent(
    model="claude-sonnet-4-5",
    guards=[
        Guard.no_pii_output(),             # Block PII in responses
        Guard.max_cost(dollars=0.50),       # Stop if cost exceeds $0.50
        Guard.no_tool_calls(["delete_*"]), # Block tools matching glob
        Guard.custom(my_check_fn),          # Your own check function
    ],
)
```

Guards run in order. First failure stops execution and raises `GuardViolationError`.

---

## Structured Output

Get typed Pydantic objects back instead of raw text:

```python
from pydantic import BaseModel

class MovieReview(BaseModel):
    title: str
    rating: float
    summary: str
    recommended: bool

agent = Agent(
    model="claude-sonnet-4-5",
    instructions="You are a movie critic.",
    output_schema=MovieReview,
)

result = agent.run("Review the movie Inception.")
review = result.output  # MovieReview instance

print(review.title)        # "Inception"
print(review.rating)       # 9.2
print(review.recommended)  # True
```

If parsing fails, Synth retries with a corrective prompt up to `max_retries` times.

---

## Pipelines

Chain agents sequentially — output of each becomes input of the next:

```python
from synth import Pipeline

researcher = Agent(model="claude-sonnet-4-5", instructions="You research topics.")
writer = Agent(model="claude-sonnet-4-5", instructions="You write clear articles.")
editor = Agent(model="claude-sonnet-4-5", instructions="You edit for clarity.")

pipeline = Pipeline([researcher, writer, editor])
result = pipeline.run("The history of the internet")
```

Run stages in parallel with `ParallelGroup`:

```python
from synth.orchestration.pipeline import ParallelGroup

pipeline = Pipeline([
    writer,
    ParallelGroup([fact_checker, style_checker]),  # Run concurrently
    editor,
])
```

Stream with stage labels:

```python
for stage_event in pipeline.stream("Write about AI"):
    print(f"[{stage_event.stage_name}] {stage_event.event}")
```

---

## Graphs

Directed-graph workflows with branching, loops, and conditional logic:

```python
from synth import Graph, node

graph = Graph()

@node(graph)
def classify(state):
    state["priority"] = "high" if "urgent" in state["text"].lower() else "low"
    return state

@node(graph)
def handle_urgent(state):
    state["response"] = "Escalating immediately."
    return state

@node(graph)
def handle_normal(state):
    state["response"] = "We'll respond within 24 hours."
    return state

graph.set_entry("classify")
graph.add_edge("classify", "handle_urgent", when=lambda s: s["priority"] == "high")
graph.add_edge("classify", "handle_normal", when=lambda s: s["priority"] == "low")
graph.add_edge("handle_urgent", Graph.END)
graph.add_edge("handle_normal", Graph.END)

result = graph.run({"text": "This is urgent! Server is down!"})
print(result.output["response"])
```

Loops are supported. Synth enforces `max_iterations=100` by default to prevent infinite loops.

Visualize your graph:

```python
print(graph.visualise())  # Outputs a Mermaid diagram
```

---

## Human-in-the-Loop

Pause a graph at specific nodes for human review before continuing:

```python
graph.with_human_in_the_loop(pause_at=["draft_email"], timeout=3600)
graph.with_checkpointing()

result = graph.run({"customer": "Alice"}, run_id="email-001")
# result is a PausedRun — inspect result.state["draft"] here

final = graph.resume("email-001", human_input="Looks good, send it.")
```

---

## Agent Teams

Coordinate multiple specialized agents under an orchestrator:

```python
from synth import AgentTeam

team = AgentTeam(
    orchestrator="claude-sonnet-4-5",
    agents=[researcher, writer, analyst],
    strategy="auto",   # orchestrator decides who does what
)

result = team.run("Write a report on renewable energy trends.")
print(result.answer)
print(result.contributions)   # Each agent's individual contribution
print(result.total_cost)
```

Use `strategy="parallel"` to run all agents concurrently.

---

## Tracing

Every run automatically records a detailed trace:

```python
result = agent.run("Summarize this document.")
trace = result.trace

print(f"Tokens: {trace.total_tokens}")
print(f"Cost: ${trace.total_cost:.4f}")
print(f"Latency: {trace.total_latency_ms:.1f}ms")

result.trace.show()                    # Open visual timeline in browser
path = result.trace.export()           # Export as OpenTelemetry JSON
```

Auto-forward all traces to an OTel collector:

```bash
export SYNTH_TRACE_ENDPOINT="https://my-otel-collector.example.com/v1/traces"
```

---

## Checkpointing

Save and resume graph execution state:

```python
graph.with_checkpointing()
result = graph.run(initial_state, run_id="my-run-001")

# Later, even in a different process
result = graph.resume("my-run-001")
```

Redis backend for distributed systems:

```python
from synth.checkpointing.redis import RedisCheckpointStore

graph.with_checkpointing(store=RedisCheckpointStore("redis://localhost:6379"))
```

---

## Evaluation

Run structured tests against your agent:

```python
from synth import Eval

evaluation = Eval(agent=agent)
evaluation.add_case(input="Capital of France?", expected="Paris")
evaluation.add_case(input="Capital of Japan?", expected="Tokyo")

report = evaluation.run()
print(f"Score: {report.overall_score}")

for case in report.cases:
    status = "PASS" if case.passed else "FAIL"
    print(f"  [{status}] {case.input} → {case.actual}")
```

Custom checker:

```python
def contains_keyword(output: str, expected: str) -> float:
    return 1.0 if expected.lower() in output.lower() else 0.0

evaluation.add_case(input="Explain photosynthesis.", expected="chlorophyll", checker=contains_keyword)
```

---

## CLI Commands

Run `synth` with no arguments to launch the interactive shell:

```bash
synth
```

```
synth> run agent.py "Hello"
synth> create agent my-bot
synth> doctor
synth> exit
```

All commands also work directly:

```bash
synth init                                  # Interactive project setup wizard
synth create agent my-bot                   # Scaffold an agent project
synth create agent my-bot -p openai         # Skip prompt, use OpenAI
synth create agentcore my-service           # AWS AgentCore project
synth create team my-team                   # Multi-agent team + pipeline
synth create tool my-tools                  # Standalone tools file
synth create mcp my-server                  # MCP server with FastMCP
synth create ui my-ui                       # Local browser testing UI
synth dev my_agent.py                       # Rich terminal UI with hot-reload
synth run my_agent.py "prompt"              # Execute agent, print result
synth bench my_agent.py "prompt" --runs 20  # Benchmark latency/cost
synth eval my_agent.py --dataset cases.json # Run evaluation suite
synth trace <run_id>                        # Open trace in browser
synth deploy --target agentcore             # Deploy to AWS AgentCore
synth deploy --target agentcore --dry-run   # Validate without deploying
synth edit agent agent.py                   # Modify existing agent config
synth doctor                                # Check env, credentials, deps
synth info --extra anthropic                # Show package info
synth help                                  # Quick reference card
```

### `synth init`

The fastest way to start a new project. Walks you through:

1. Project name and description
2. Provider selection
3. Model selection (region-aware for AgentCore with Bedrock model catalog)
4. Agent instructions
5. **Tool Wizard** — pick pre-built tools or scaffold custom `@tool` stubs
6. **MCP Wizard** — pick pre-built MCP servers or scaffold custom `@mcp.tool()` stubs
7. Feature toggles (memory, guards, structured output, eval, deploy)
8. Credential check (AgentCore only)
9. Summary and confirmation
10. Project generation
11. Optional "Deploy now?" prompt (AgentCore only)

For AgentCore projects, `synth init` also:
- Auto-detects AWS credentials (env vars → `~/.aws/credentials` → AWS Toolkit profiles)
- Prompts for target AWS region (default: `us-east-1`)
- Shows Bedrock models available in that region
- Writes `aws_region`, `model_id`, `cris_enabled`, and `aws_profile` to `agentcore.yaml`

### `synth dev`

Rich terminal UI for interactive development:

```bash
synth dev my_agent.py
```

Features: streaming token-by-token output, tool call visualization, slash commands (`/tools`, `/reload`, `/trace`, `/export`, `/clear`, `/cost`, `/quit`), markdown rendering, status bar with live cost/token tracking.

### `synth deploy`

Guided six-stage deployment wizard:

```bash
synth deploy --target agentcore my_agent.py
synth deploy --target agentcore --dry-run my_agent.py  # Stages 1–4 only
```

Stages: credential validation → dependency check → file validation → manifest generation → artifact packaging → AgentCore API submission. Each prints `[  OK  ]` or `[FAIL]` with a corrective suggestion on failure.

### `synth edit agent`

Interactively modify an existing agent without editing files manually:

```bash
synth edit agent agent.py
```

Menu options: (a) instructions, (b) model, (c) tools, (d) MCP servers. Shows a diff before writing. Uses atomic temp-file rename to prevent corruption.

### `synth doctor`

```bash
synth doctor
```

Checks: Python version, core dependencies, provider API keys, `SYNTH_TRACE_ENDPOINT` format, optional provider packages, and (when `agentcore.yaml` is present) AgentCore config fields (`aws_region`, `model_id`, `cris_enabled`, `aws_profile`).

### `synth bench`

```bash
synth bench my_agent.py "Hello" --runs 20 --warmup 2
```

Reports p50/p95/p99 latency, average tokens, cost per run, and success rate.

---

## Deploying to AWS AgentCore

### Prerequisites

```bash
pip install synth-agent-sdk[agentcore]
```

### Wrapping Your Agent

```python
from synth import Agent
from synth.deploy.agentcore import agentcore_handler

agent = Agent(
    model="bedrock/claude-sonnet-4-5",
    instructions="You are a customer support agent.",
    tools=[lookup_order, check_inventory],
)

app = agentcore_handler(agent)
```

### Deploy

```bash
synth deploy --target agentcore --dry-run   # Validate first
synth deploy --target agentcore             # Deploy
```

The packager automatically excludes `.env` files, credential files, and `.synth/checkpoints/` from the artifact. It also scans `agentcore.yaml` for accidental credential patterns and aborts if any are found.

### Secure User Identity

```python
from synth.deploy.agentcore import extract_user_id

user_id = extract_user_id(context)  # Extracts from signed JWT in RequestContext
```

### Gateway MCP Client

```python
from synth.deploy.agentcore import create_gateway_client

client = create_gateway_client(
    gateway_url="https://my-gateway.example.com",
    client_id_param="/myapp/gateway/client_id",
    client_secret_param="/myapp/gateway/client_secret",
)
mcp_client = client.as_mcp_client()
```

### Code Interpreter

```python
from synth.deploy.agentcore import CodeInterpreterTools

ci = CodeInterpreterTools()
result = ci.execute_python("import math; print(math.sqrt(144))")
print(result)  # "12.0"
```

### SSM Config

```python
from synth.deploy.agentcore import get_ssm_parameter

db_url = get_ssm_parameter("/myapp/prod/db_url")
api_key = get_ssm_parameter("/myapp/prod/api_key", decrypt=True)
```

---

## Error Handling

All Synth errors inherit from `SynthError` and include `component` and `suggestion` fields.

| Error | When |
|-------|------|
| `SynthConfigError` | Missing API key, invalid model, missing provider package |
| `ToolDefinitionError` | `@tool` missing type annotations or docstring |
| `ToolExecutionError` | Tool function raised an exception |
| `GuardViolationError` | A guard check failed |
| `CostLimitError` | Cost guard limit exceeded |
| `SynthParseError` | Structured output couldn't be parsed after retries |
| `GraphRoutingError` | No edge condition matched at a graph node |
| `GraphLoopError` | Graph exceeded `max_iterations` |
| `RunNotFoundError` | No checkpoint found for the given `run_id` |
| `PipelineError` | A pipeline stage failed |

```python
from synth.errors import SynthConfigError, ToolExecutionError, GuardViolationError

try:
    result = agent.run("Do something risky.")
except GuardViolationError as e:
    print(f"Guard '{e.guard_name}' blocked: {e.remediation}")
except ToolExecutionError as e:
    print(f"Tool '{e.tool_name}' failed: {e.original_error}")
except SynthConfigError as e:
    print(f"Config issue in {e.component}: {e.suggestion}")
```

---

## Environment Variables

| Variable | Purpose | Required? |
|----------|---------|-----------|
| `ANTHROPIC_API_KEY` | Anthropic Claude API key | Only for `claude-*` models |
| `OPENAI_API_KEY` | OpenAI GPT API key | Only for `gpt-*` models |
| `GOOGLE_API_KEY` | Google Gemini API key | Only for `gemini-*` models |
| `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY` | AWS credentials for Bedrock | Only for `bedrock/*` (or use IAM) |
| `SYNTH_TRACE_ENDPOINT` | HTTPS URL of an OTel collector | No |
| `SYNTH_NO_BANNER` | Set to `1` to skip the boot sequence | No |
| `NO_COLOR` | Disable colored terminal output | No |

---

## FAQ

**Do I need an API key?**
Yes, for cloud models. Ollama runs locally and needs no key.

**Can I use Synth in Jupyter?**
Yes. Synth detects an existing event loop and handles it automatically.

**How do I switch models?**
Change the `model` string. Install the matching extra and set the API key.

**What if the provider is down?**
Synth retries on HTTP 429 and 5xx with exponential backoff. Configure with `max_retries` and `retry_backoff`.

**Can I use multiple models in one app?**
Yes. Each `Agent` has its own model.

**How do I debug what my agent is doing?**
Use `result.trace.show()` for a visual timeline, or `synth dev my_agent.py` for an interactive terminal UI with `/trace` command.

**Is my data secure?**
Synth never logs or serializes API keys. Guards run before side-effecting operations. Checkpoints use JSON only. All provider calls use HTTPS.

**What are the core dependencies?**
`pydantic`, `httpx`, `click`, `typing-extensions`, `rich`, `prompt-toolkit`. Provider SDKs are optional extras.

---

## License

MIT