# AgentKit (LangChain Agent Runtime)

LangChain's production agent runtime: graph-based, streaming, checkpointing, horizontal scaling.

---

## Overview

AgentKit is LangChain's **agent runtime** — the execution layer for LangGraph graphs in production. It provides:

- **Horizontal scaling** — Multiple workers, shared state
- **Streaming** — Token-level, structured, custom
- **Checkpointing** — Postgres, SQLite, Redis
- **Multi-tenancy** — Thread isolation, auth
- **Observability** — LangSmith native
- **API** — REST, WebSocket, GraphQL

---

## Architecture

```
┌─────────────────────────────────────┐
│           AgentKit Server           │
├─────────────────────────────────────┤
│  API Layer (REST/WS/GraphQL)        │
├─────────────────────────────────────┤
│  Execution Engine                   │
│  ├── Graph Executor                 │
│  ├── Stream Manager                 │
│  ├── Checkpoint Coordinator         │
│  └── Tenant Manager                 │
├─────────────────────────────────────┤
│  State Backend (Postgres/Redis)     │
├─────────────────────────────────────┤
│  LangSmith (Observability)          │
└─────────────────────────────────────┘
```

---

## Quick Start

### 1. Define Graph (LangGraph)
```python
# agent.py
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class State(TypedDict):
    messages: Annotated[list, operator.add]
    user_id: str

def reasoning(state):
    # Your logic
    return {"messages": [response]}

graph = StateGraph(State)
graph.add_node("reasoning", reasoning)
graph.add_edge(START, "reasoning")
graph.add_edge("reasoning", END)

compiled = graph.compile()
```

### 2. Configure AgentKit
```yaml
# agentkit.yaml
server:
  host: "0.0.0.0"
  port: 8000
  workers: 4

database:
  url: "postgresql://user:pass@localhost/agentkit"
  pool_size: 20

redis:
  url: "redis://localhost:6379/0"

langsmith:
  api_key: "${LANGSMITH_API_KEY}"
  project: "my-agents"

auth:
  enabled: true
  jwt_secret: "${JWT_SECRET}"
  allowed_origins: ["https://app.example.com"]
```

### 3. Run
```bash
# Install
pip install agentkit

# Run server
agentkit serve --config agentkit.yaml

# Or programmatically
from agentkit import AgentKitServer

server = AgentKitServer.from_config("agentkit.yaml")
server.add_graph("my_agent", compiled)
server.run()
```

---

## API Reference

### REST Endpoints
| Endpoint | Method | Description |
|---|---|---|
| `/threads` | POST | Create thread |
| `/threads/{thread_id}` | GET | Get thread state |
| `/threads/{thread_id}/runs` | POST | Start run |
| `/threads/{thread_id}/runs/{run_id}` | GET | Get run status |
| `/threads/{thread_id}/runs/{run_id}/stream` | GET | Stream run |
| `/threads/{thread_id}/state` | POST | Update state (fork) |
| `/threads/{thread_id}/history` | GET | State history |

### Create Thread
```bash
curl -X POST http://localhost:8000/threads \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{"graph_id": "my_agent", "metadata": {"user_id": "123"}}'
```

### Run Graph
```bash
curl -X POST http://localhost:8000/threads/$THREAD_ID/runs \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{"input": {"messages": [{"role": "user", "content": "Hello"}]}}'
```

### Stream
```bash
curl -N http://localhost:8000/threads/$THREAD_ID/runs/$RUN_ID/stream \
  -H "Authorization: Bearer $TOKEN"
# SSE stream: data: {"type": "token", "data": "..."}
```

---

## Multi-Tenancy

### Thread Isolation
```python
# Each thread = isolated conversation
thread_config = {
    "configurable": {
        "thread_id": "tenant-123:session-456",
        "user_id": "user-789"
    }
}

# State scoped to thread
state = await server.get_thread_state(thread_id)
```

### Authentication
```python
# JWT-based
# Header: Authorization: Bearer <jwt>
# JWT claims: { "sub": "user-123", "tenant": "acme-corp", "roles": ["user"] }

# API Key
# Header: X-API-Key: ak_...
```

### Tenant Configuration
```yaml
# agentkit.yaml
tenants:
  acme-corp:
    rate_limit: 1000  # req/min
    max_threads: 10000
    max_run_time: 300  # seconds
    allowed_graphs: ["support_agent", "sales_agent"]
    pii_masking: true
```

---

## Checkpointing

### Backends
| Backend | Config | Use Case |
|---|---|---|
| **PostgreSQL** | `postgresql://...` | Production, durability |
| **SQLite** | `sqlite:///checkpoints.db` | Dev, single-node |
| **Redis** | `redis://...` | High-throughput, ephemeral |

### Configuration
```yaml
checkpointer:
  type: "postgres"
  url: "postgresql://user:pass@localhost/agentkit"
  pool_size: 20
  # Table: checkpoint_writes, checkpoint_blobs
  # Auto-migrates schema
```

### Migration
```python
# Schema versioning automatic
# Custom migration
from agentkit.migrations import migrate

migrate(
    from_version=1,
    to_version=2,
    migration_fn=lambda state: {
        **state,
        "schema_version": 2,
        "new_field": "default"
    }
)
```

---

## Streaming

### Stream Modes
```python
# In API request
{
    "input": {...},
    "stream_mode": "messages"  # or "values", "updates", "debug"
}
```

### Server-Sent Events (SSE)
```javascript
// Client
const evtSource = new EventSource(`/threads/${threadId}/runs/${runId}/stream`, {
  headers: { Authorization: `Bearer ${token}` }
});

evtSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  if (data.type === "token") {
    appendToUI(data.data);
  }
};
```

### Custom Streaming
```python
# In graph node
def streaming_node(state):
    for token in llm.stream(state["prompt"]):
        yield {"custom_event": token}  # Custom stream mode
```

---

## Scaling

### Horizontal Scaling
```yaml
# agentkit.yaml
server:
  workers: 4  # Processes
  threads_per_worker: 100

# Behind load balancer
# Sticky sessions NOT required (stateless workers)
# State in Postgres/Redis
```

### Queue Management
```yaml
queue:
  max_size: 10000
  timeout: 300  # Max wait time
  priority_levels: 3
  dead_letter_ttl: 86400
```

### Rate Limiting
```yaml
rate_limit:
  global: 10000  # req/min
  per_tenant: 1000
  per_user: 100
  burst: 50
```

---

## Observability

### LangSmith Integration
```yaml
langsmith:
  api_key: "${LANGSMITH_API_KEY}"
  project: "production-agents"
  tags: ["production", "v2.1"]
  metadata:
    environment: "production"
    region: "us-east-1"
```

### Custom Metrics
```python
from agentkit.metrics import MetricsCollector

metrics = MetricsCollector()

@metrics.track("agent_runtime")
async def my_node(state):
    # Automatically tracks: latency, tokens, errors
    return result
```

### Health Checks
```bash
# Liveness
curl http://localhost:8000/health/live

# Readiness
curl http://localhost:8000/health/ready
# Checks: DB connection, Redis, graph compilation
```

---

## Deployment

### Docker
```dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY agent.py agentkit.yaml .
CMD ["agentkit", "serve", "--config", "agentkit.yaml"]
```

### Kubernetes
```yaml
# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agentkit
spec:
  replicas: 4
  selector:
    matchLabels:
      app: agentkit
  template:
    spec:
      containers:
      - name: agentkit
        image: myorg/agentkit:latest
        ports:
        - containerPort: 8000
        envFrom:
        - secretRef:
            name: agentkit-secrets
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "2Gi"
            cpu: "2000m"
---
# Service
apiVersion: v1
kind: Service
metadata:
  name: agentkit
spec:
  selector:
    app: agentkit
  ports:
  - port: 80
    targetPort: 8000
```

### LangGraph Platform (Managed)
```bash
# Zero-config managed AgentKit
langgraph deploy --project my-agents
# Auto-scaling, managed Postgres, LangSmith, auth
```

---

## Migration from LangGraph Direct

| Before (Direct) | After (AgentKit) |
|---|---|
| `graph.compile(checkpointer=PostgresSaver())` | Config in `agentkit.yaml` |
| Manual thread management | `/threads` API |
| Custom streaming endpoint | Built-in SSE |
| Manual auth | JWT/API key built-in |
| Custom metrics | LangSmith + custom |
| Single process | Horizontal scaling |

---

## Related Notes

- [[Frameworks/LangGraph.md]] — Graph definition
- [[Frameworks/Framework-Comparison.md]] — vs others
- [[StateManagement/Checkpointing.md]] — Checkpointing patterns
- [[Production/Observability.md]] — LangSmith
- [[Production/Deployment-Architecture.md]] — K8s, scaling
- [[ContextEngineering/Memory-Systems.md]] — Thread-scoped memory
- [[Production/Security.md]] — Multi-tenancy, auth

---

*Sources: AgentKit docs (langchain-ai.github.io/agentkit), LangGraph Platform docs, LangChain blog, AgentKit GitHub, production deployment guides.*