Raw SDK: Anthropic and OpenAI APIs from Scratch

After this lesson, you will be able to: Call AI APIs directly, implement tool use, streaming, function calling, and structured outputs without any framework.

Frameworks are great until they aren't. When you need cost control, latency optimization, or behavior the framework doesn't expose, you drop to the raw SDK. This lesson covers the patterns frameworks abstract: streaming, structured outputs, parallel tool calls, prompt caching.

Prerequisites:CrewAI: Multi-Agent Teams with Roles and Tasks

Streaming responses

Stream tokens as they arrive instead of waiting for the full response.

python

import anthropic

client = anthropic.Anthropic()
with client.messages.stream(
    model='claude-sonnet-4-6',
    max_tokens=1024,
    messages=[{'role': 'user', 'content': 'Write a short poem about the sea.'}],
) as stream:
    for text in stream.text_stream:
        print(text, end='', flush=True)
print()  # newline at end

Structured outputs (Anthropic)

Force the model to return JSON matching a schema, via tool use.

python

import anthropic, json

schema = {
    'name': 'extract_person',
    'description': 'Extract a person from text.',
    'input_schema': {
        'type': 'object',
        'properties': {
            'name': {'type': 'string'},
            'age': {'type': 'integer'},
            'role': {'type': 'string'},
        },
        'required': ['name', 'age', 'role'],
    },
}
client = anthropic.Anthropic()
resp = client.messages.create(
    model='claude-sonnet-4-6', max_tokens=512, tools=[schema], tool_choice={'type':'tool','name':'extract_person'},
    messages=[{'role':'user','content':'Maria is 34 and runs a bakery.'}],
)
data = next(b for b in resp.content if b.type == 'tool_use').input
print(json.dumps(data, indent=2))

Prompt caching

If you reuse a long system prompt or document across many turns, mark it as cached. Claude charges 90% less for cached content. Add `cache_control: {'type': 'ephemeral'}` to a content block. Lifesaver for RAG and code-context apps.

Async + concurrent agent calls

asyncio = throughput. Critical for production.

python

import asyncio, anthropic

client = anthropic.AsyncAnthropic()

async def ask(question):
    resp = await client.messages.create(
        model='claude-3-5-haiku-latest', max_tokens=200,
        messages=[{'role': 'user', 'content': question}],
    )
    return resp.content[0].text

async def main():
    questions = ['Capital of France?', 'Capital of Japan?', 'Capital of Egypt?']
    results = await asyncio.gather(*[ask(q) for q in questions])
    for q, a in zip(questions, results):
        print(f'{q} → {a}')

asyncio.run(main())

←CrewAI: Multi-Agent Teams with Roles and Tasks

Back to Code-First Agents

Model Context Protocol (MCP): Connect Agents to External Tools→