How to Build a RAG Pipeline in 100 Lines of Python
Retrieval-Augmented Generation is the most practical AI pattern of 2025. Here's a minimal but production-ready implementation using LangChain, ChromaDB, and the OpenAI API.
Claude Code is no longer just a terminal tool — it's a full agentic API. This tutorial shows you how to go from your first API call to building autonomous coding agents in Python or TypeScript.
Muunsparks
2026-03-17

Anthropic's Claude Code has evolved far beyond a terminal coding assistant. With the Claude Agent SDK, you can now embed Claude Code's full agentic capabilities — file reading, shell execution, web search, and more — directly into your own applications. This post walks you through everything you need to get up and running.
There are two ways developers commonly interact with Claude programmatically:
https://api.anthropic.com/v1/messages for sending prompts and getting responses. Great for chatbots, text generation, and tool-use pipelines you control manually.This tutorial covers both, starting with the simpler Messages API before moving on to the Agent SDK.
Set your API key as an environment variable so you never hard-code it:
export ANTHROPIC_API_KEY="sk-ant-..."
The Messages API is your starting point for any Claude integration. It's a simple POST request.
Python:
pip install anthropic
TypeScript / Node.js:
npm install @anthropic-ai/sdk
Python:
import anthropic
client = anthropic.Anthropic() # Reads ANTHROPIC_API_KEY from environment
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{"role": "user", "content": "Explain async/await in Python in two sentences."}
],
)
print(message.content[0].text)
TypeScript:
import Anthropic from "@anthropic-ai/sdk";
const client = new Anthropic(); // Reads ANTHROPIC_API_KEY from environment
const message = await client.messages.create({
model: "claude-sonnet-4-6",
max_tokens: 1024,
messages: [
{
role: "user",
content: "Explain async/await in Python in two sentences.",
},
],
});
console.log(message.content[0].text);
Or via raw cURL:
curl https://api.anthropic.com/v1/messages \
--header "x-api-key: $ANTHROPIC_API_KEY" \
--header "anthropic-version: 2023-06-01" \
--header "content-type: application/json" \
--data '{
"model": "claude-sonnet-4-6",
"max_tokens": 1024,
"messages": [{"role": "user", "content": "Hello, Claude"}]
}'
| Model | Best For |
|---|---|
| claude-opus-4-6 | Complex reasoning, deep analysis, difficult coding |
| claude-sonnet-4-6 | Balanced intelligence and speed — ideal for most production workloads |
| claude-haiku-4-5 | Fast, lightweight tasks at lower cost |
For a better user experience (especially with longer outputs), use streaming:
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "Write me a sorting algorithm in Python"}],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
Claude can call tools you define. Here's a minimal example:
tools = [
{
"name": "get_file_contents",
"description": "Read the contents of a file by path",
"input_schema": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "The file path to read"}
},
"required": ["path"],
},
}
]
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
tools=tools,
messages=[{"role": "user", "content": "What's in my README.md?"}],
)
# Check if Claude wants to call a tool
if response.stop_reason == "tool_use":
for block in response.content:
if block.type == "tool_use":
print(f"Tool called: {block.name}")
print(f"With input: {block.input}")
The Agent SDK is where things get really powerful. Instead of managing the tool loop yourself, you delegate an entire task to Claude and let the SDK handle the back-and-forth autonomously.
Python:
pip install claude-agent-sdk
TypeScript / Node.js:
npm install @anthropic-ai/claude-code
Note: The Claude Code CLI is automatically bundled with the Python package — no separate installation required.
The query() function is an async generator that yields messages as Claude works through your task:
Python:
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock
async def main():
async for message in query(prompt="List all Python files in this directory"):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text)
asyncio.run(main())
TypeScript:
import { query, ClaudeAgentOptions } from "@anthropic-ai/claude-code";
for await (const message of query({
prompt: "List all Python files in this directory",
})) {
if (message.type === "assistant") {
for (const block of message.content) {
if (block.type === "text") {
console.log(block.text);
}
}
}
}
Use ClaudeAgentOptions to control Claude's behavior:
options = ClaudeAgentOptions(
system_prompt="You are a senior Python engineer. Be concise and precise.",
max_turns=10,
cwd="/path/to/my/project", # Set the working directory
)
async for message in query(
prompt="Refactor the utils.py file to use dataclasses",
options=options
):
print(message)
By default, Claude has access to its full toolset: Read, Write, Edit, Bash, and more. You can restrict or auto-approve tools:
options = ClaudeAgentOptions(
allowed_tools=["Read", "Glob"], # Auto-approve these — Claude won't ask for confirmation
disallowed_tools=["Bash"], # Block these entirely
permission_mode="acceptEdits", # Auto-accept file edits
)
async for message in query(
prompt="Find all TODO comments in the codebase",
options=options
):
pass
Permission modes:
| Mode | Behavior |
|---|---|
| default | Claude asks before each sensitive action |
| acceptEdits | Auto-approves file edits, still asks for shell commands |
| bypassPermissions | Full auto-approval (use only in trusted environments) |
One of the Agent SDK's most powerful features is the ability to define your own Python functions as tools that Claude can invoke:
from claude_agent_sdk import ClaudeSDKClient
async def send_slack_notification(channel: str, message: str) -> str:
"""Send a Slack notification to a channel."""
# Your Slack integration here
return f"Sent to #{channel}: {message}"
async def main():
client = ClaudeSDKClient(
custom_tools=[send_slack_notification]
)
async with client:
response = await client.query(
"Analyze the test results in test_output.log and notify #engineering on Slack with a summary"
)
print(response)
asyncio.run(main())
Custom tools run as in-process MCP servers — no separate process or network setup required.
For high-volume, non-time-sensitive workloads, the Message Batches API processes requests asynchronously at a 50% cost reduction.
batch = client.messages.batches.create(
requests=[
{
"custom_id": "review-pr-1",
"params": {
"model": "claude-sonnet-4-6",
"max_tokens": 1024,
"messages": [{"role": "user", "content": "Review this diff: ..."}],
},
},
{
"custom_id": "review-pr-2",
"params": {
"model": "claude-sonnet-4-6",
"max_tokens": 1024,
"messages": [{"role": "user", "content": "Review this diff: ..."}],
},
},
]
)
print(f"Batch ID: {batch.id}")
# Poll batch.processing_status until "ended"
Putting it all together: an agent that reviews a pull request, writes a summary, and saves it to a file.
import asyncio
from pathlib import Path
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock
async def review_pr(diff_path: str, output_path: str):
options = ClaudeAgentOptions(
system_prompt="""You are an expert code reviewer.
Analyze the provided diff for: correctness, security issues,
performance concerns, and code style. Be specific and actionable.""",
cwd=str(Path(diff_path).parent),
allowed_tools=["Read", "Write"],
permission_mode="acceptEdits",
max_turns=5,
)
prompt = f"""
Please review the git diff at {diff_path}.
Write a structured code review to {output_path} with sections for:
- Summary of changes
- Issues found (critical / minor)
- Suggestions for improvement
- Overall assessment
"""
print("🔍 Starting code review...")
async for message in query(prompt=prompt, options=options):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock) and block.text.strip():
print(f" → {block.text[:80]}...")
print(f"✅ Review saved to {output_path}")
asyncio.run(review_pr("changes.diff", "review.md"))
Messages API — direct request/response, you manage state and tool loops manually. Best for: chatbots, simple Q&A, tool-use pipelines.
Agent SDK — autonomous task execution with Claude managing the full loop. Best for: complex multi-step coding tasks, file manipulation, automated workflows.
Batch API — async bulk processing at reduced cost. Best for: large-scale analysis, nightly jobs, non-urgent workloads.
Happy building! If you run into issues, the Anthropic Discord and support center are great places to get help.
// RELATED ARTICLES
Retrieval-Augmented Generation is the most practical AI pattern of 2025. Here's a minimal but production-ready implementation using LangChain, ChromaDB, and the OpenAI API.
ChatGPT isn't a money printer. But used strategically, it can cut your time-to-output in half. Here's how people are actually monetizing it.
OpenAI's market share dropped from 55% to 40% in twelve months. DeepSeek trains for $6M what costs others $100M+. The model layer is commoditizing.