import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
prompt: "Use the docs MCP server to explain what hooks are in Claude Code",
options: {
mcpServers: {
"claude-code-docs": {
type: "http",
url: "https://code.claude.com/docs/mcp"
}
},
allowedTools: ["mcp__claude-code-docs__*"]
}
})) {
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage


async def main():
options = ClaudeAgentOptions(
mcp_servers={
"claude-code-docs": {
"type": "http",
"url": "https://code.claude.com/docs/mcp",
}
},
allowed_tools=["mcp__claude-code-docs__*"],
)

async for message in query(
prompt="Use the docs MCP server to explain what hooks are in Claude Code",
options=options,
):
if isinstance(message, ResultMessage) and message.subtype == "success":
print(message.result)


asyncio.run(main())

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
prompt: "List files in my project",
options: {
mcpServers: {
filesystem: {
command: "npx",
args: ["-y", "@modelcontextprotocol/server-filesystem", "/Users/me/projects"]
}
},
allowedTools: ["mcp__filesystem__*"]
}
})) {
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage


async def main():
options = ClaudeAgentOptions(
mcp_servers={
"filesystem": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"/Users/me/projects",
],
}
},
allowed_tools=["mcp__filesystem__*"],
)

async for message in query(prompt="List files in my project", options=options):
if isinstance(message, ResultMessage) and message.subtype == "success":
print(message.result)


asyncio.run(main())

const _ = {
options: {
mcpServers: {
// your servers
},
allowedTools: [
"mcp__github__*", // All tools from the github server
"mcp__db__query", // Only the query tool from db server
"mcp__slack__send_message" // Only send_message from slack server
]
}
};

options = ClaudeAgentOptions(
mcp_servers={
# your servers
},
allowed_tools=[
"mcp__github__*",  # All tools from the github server
"mcp__db__query",  # Only the query tool from db server
"mcp__slack__send_message",  # Only send_message from slack server
],
)

import { query } from "@anthropic-ai/claude-agent-sdk";

const options = {
mcpServers: {
// your servers
},
};

for await (const message of query({ prompt: "...", options })) {
if (message.type === "system" && message.subtype === "init") {
const mcpTools = message.tools.filter((name) => name.startsWith("mcp__"));
console.log("Available MCP tools:", mcpTools);
}
}

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, SystemMessage


async def main():
options = ClaudeAgentOptions(
mcp_servers={
# your servers
},
)
async for message in query(prompt="...", options=options):
if isinstance(message, SystemMessage) and message.subtype == "init":
mcp_tools = [t for t in message.data.get("tools", []) if t.startswith("mcp__")]
print("Available MCP tools:", mcp_tools)


asyncio.run(main())

const _ = {
options: {
mcpServers: {
filesystem: {
command: "npx",
args: ["-y", "@modelcontextprotocol/server-filesystem", "/Users/me/projects"]
}
},
allowedTools: ["mcp__filesystem__read_file", "mcp__filesystem__list_directory"]
}
};

options = ClaudeAgentOptions(
mcp_servers={
"filesystem": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"/Users/me/projects",
],
}
},
allowed_tools=["mcp__filesystem__read_file", "mcp__filesystem__list_directory"],
)

const _ = {
options: {
mcpServers: {
"remote-api": {
type: "sse",
url: "https://api.example.com/mcp/sse",
headers: {
Authorization: `Bearer ${process.env.API_TOKEN}`
}
}
},
allowedTools: ["mcp__remote-api__*"]
}
};

options = ClaudeAgentOptions(
mcp_servers={
"remote-api": {
"type": "sse",
"url": "https://api.example.com/mcp/sse",
"headers": {"Authorization": f"Bearer {os.environ['API_TOKEN']}"},
}
},
allowed_tools=["mcp__remote-api__*"],
)

// After completing OAuth flow in your app.
// Implement getAccessTokenFromOAuthFlow for your OAuth provider.
const accessToken = await getAccessTokenFromOAuthFlow();

const options = {
mcpServers: {
"oauth-api": {
type: "http",
url: "https://api.example.com/mcp",
headers: {
Authorization: `Bearer ${accessToken}`
}
}
},
allowedTools: ["mcp__oauth-api__*"]
};

# After completing OAuth flow in your app.
# Implement get_access_token_from_oauth_flow for your OAuth provider.
access_token = await get_access_token_from_oauth_flow()

options = ClaudeAgentOptions(
mcp_servers={
"oauth-api": {
"type": "http",
"url": "https://api.example.com/mcp",
"headers": {"Authorization": f"Bearer {access_token}"},
}
},
allowed_tools=["mcp__oauth-api__*"],
)

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
prompt: "List the 3 most recent issues in anthropics/claude-code",
options: {
mcpServers: {
github: {
type: "http",
url: "https://api.githubcopilot.com/mcp/",
headers: {
Authorization: `Bearer ${process.env.GITHUB_TOKEN}`
}
}
},
allowedTools: ["mcp__github__list_issues"]
}
})) {
// Verify MCP server connected successfully
if (message.type === "system" && message.subtype === "init") {
console.log("MCP servers:", message.mcp_servers);
}

// Log when Claude calls an MCP tool
if (message.type === "assistant") {
for (const block of message.message.content) {
if (block.type === "tool_use" && block.name.startsWith("mcp__")) {
console.log("MCP tool called:", block.name);
}
}
}

// Print the final result
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}

import asyncio
import os
from claude_agent_sdk import (
query,
ClaudeAgentOptions,
ResultMessage,
SystemMessage,
AssistantMessage,
)


async def main():
options = ClaudeAgentOptions(
mcp_servers={
"github": {
"type": "http",
"url": "https://api.githubcopilot.com/mcp/",
"headers": {"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}"},
}
},
allowed_tools=["mcp__github__list_issues"],
)

async for message in query(
prompt="List the 3 most recent issues in anthropics/claude-code",
options=options,
):
# Verify MCP server connected successfully
if isinstance(message, SystemMessage) and message.subtype == "init":
print("MCP servers:", message.data.get("mcp_servers"))

# Log when Claude calls an MCP tool
if isinstance(message, AssistantMessage):
for block in message.content:
if hasattr(block, "name") and block.name.startswith("mcp__"):
print("MCP tool called:", block.name)

# Print the final result
if isinstance(message, ResultMessage) and message.subtype == "success":
print(message.result)


asyncio.run(main())

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
// Natural language query - Claude writes the SQL
prompt: "How many users signed up last week? Break it down by day.",
options: {
mcpServers: {
postgres: {
command: "npx",
// dbhub.toml sets readonly = true, so execute_sql rejects writes
args: ["-y", "@bytebase/dbhub", "--config", "dbhub.toml"]
}
},
allowedTools: ["mcp__postgres__execute_sql"]
}
})) {
if (message.type === "result" && message.subtype === "success") {
console.log(message.result);
}
}

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage


async def main():
options = ClaudeAgentOptions(
mcp_servers={
"postgres": {
"command": "npx",
# dbhub.toml sets readonly = true, so execute_sql rejects writes
"args": [
"-y",
"@bytebase/dbhub",
"--config",
"dbhub.toml",
],
}
},
allowed_tools=["mcp__postgres__execute_sql"],
)

# Natural language query - Claude writes the SQL
async for message in query(
prompt="How many users signed up last week? Break it down by day.",
options=options,
):
if isinstance(message, ResultMessage) and message.subtype == "success":
print(message.result)


asyncio.run(main())

import { query } from "@anthropic-ai/claude-agent-sdk";

try {
for await (const message of query({
prompt: "Process data",
options: {
mcpServers: {
// Replace dataServer with your server configuration
"data-processor": dataServer
}
}
})) {
if (message.type === "system" && message.subtype === "init") {
const unavailableServers = message.mcp_servers.filter(
(s) => s.status === "failed" || s.status === "needs-auth"
);

if (unavailableServers.length > 0) {
console.warn("Unavailable MCP servers:", unavailableServers);
}
}

if (message.type === "result" && message.subtype === "error_during_execution") {
console.error("Execution failed");
}
}
} catch (error) {
// A single-shot query() throws after yielding an error result. If the
// failure was an error result, the error subtype branch above has
// already run; a failure to start or reach the Claude Code process
// yields no result message. MCP servers that fail to connect don't
// throw: use the status check above, and note that servers still
// "pending" at init need a later status check.
console.log(`Session ended with an error: ${error}`);
}

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, SystemMessage, ResultMessage


async def main():
# Replace data_server with your server configuration
options = ClaudeAgentOptions(mcp_servers={"data-processor": data_server})

try:
async for message in query(prompt="Process data", options=options):
if isinstance(message, SystemMessage) and message.subtype == "init":
unavailable_servers = [
s
for s in message.data.get("mcp_servers", [])
if s.get("status") in ("failed", "needs-auth")
]

if unavailable_servers:
print(f"Unavailable MCP servers: {unavailable_servers}")

if (
isinstance(message, ResultMessage)
and message.subtype == "error_during_execution"
):
print("Execution failed")
except Exception as error:
# A single-shot query() raises after yielding an error result. If the
# failure was an error result, the error subtype branch above has
# already run; a failure to start or reach the Claude Code process
# yields no result message. MCP servers that fail to connect don't
# raise: use the status check above, and note that servers still
# "pending" at init need a later status check.
print(f"Session ended with an error: {error}")


asyncio.run(main())

if (message.type === "system" && message.subtype === "init") {
for (const server of message.mcp_servers) {
if (server.status === "failed") {
console.error(`Server ${server.name} failed to connect`);
}
}
}

if isinstance(message, SystemMessage) and message.subtype == "init":
for server in message.data.get("mcp_servers", []):
if server.get("status") == "failed":
print(f"Server {server['name']} failed to connect")

const _ = {
options: {
mcpServers: {
// your servers
},
allowedTools: ["mcp__servername__*"] // Auto-approve calls from this server
}
};

options = ClaudeAgentOptions(
mcp_servers={
# your servers
},
allowed_tools=["mcp__servername__*"],  # Auto-approve calls from this server
)