hawk-sdk

Python SDK for the Hawk daemon API.

Installation

pip install hawk-sdk

Quick Start

from hawk import HawkClient

with HawkClient() as client:
    # Check health
    health = client.health()
    print(f"Status: {health.status}, Version: {health.version}")

    # Chat
    response = client.chat("Explain async/await in Python")
    print(response.response)

Async Usage

import asyncio
from hawk import AsyncHawkClient

async def main():
    async with AsyncHawkClient() as client:
        response = await client.chat("Hello!")
        print(response.response)

asyncio.run(main())

Streaming

from hawk import HawkClient

with HawkClient() as client:
    with client.chat_stream("Write a haiku") as stream:
        for event in stream.events():
            print(event.data, end="", flush=True)
        print()

Or collect the full text:

with HawkClient() as client:
    with client.chat_stream("Write a haiku") as stream:
        text = stream.collect_text()
        print(text)

Tools

from hawk import HawkClient, Tool, tool, chat_with_tools

@tool(
    name="get_weather",
    description="Get current weather for a location",
    parameters={
        "type": "object",
        "properties": {"location": {"type": "string"}},
        "required": ["location"],
    },
)
def get_weather(location: str) -> str:
    return f"Sunny, 72F in {location}"

with HawkClient() as client:
    response = chat_with_tools(
        client,
        "What's the weather in NYC?",
        tools=[get_weather],
    )
    print(response.response)

Workflow

from hawk import Workflow
from hawk.retry import RetryConfig

def fetch(url: str) -> str:
    import httpx
    return httpx.get(url).text

def summarize(text: str) -> str:
    # Process the text
    return text[:100]

wf = (
    Workflow("fetch-and-summarize")
    .step("fetch", fetch, timeout=10.0)
    .step("summarize", summarize, retry=RetryConfig(max_retries=2))
    .build()
)

result = wf.run("https://example.com")

Agent

from hawk import HawkClient, Agent, AgentConfig, Tool

weather_tool = Tool(
    name="get_weather",
    description="Get weather",
    parameters={"type": "object", "properties": {"location": {"type": "string"}}},
    fn=lambda location: f"72F in {location}",
)

with HawkClient() as client:
    agent = Agent(client, AgentConfig(
        name="weather-bot",
        model="claude-sonnet-4-20250514",
        tools=[weather_tool],
    ))

    response = agent.chat("What's the weather in San Francisco?")
    print(response.response)

    # Conversation history is maintained
    response = agent.chat("What about New York?")
    print(response.response)

Sessions

from hawk import HawkClient

with HawkClient() as client:
    # List sessions
    sessions = client.list_sessions(limit=10)
    for s in sessions.data:
        print(f"{s.id}: {s.turns} turns")

    # Get session details
    detail = client.get_session("session-id")
    print(f"Model: {detail.model}, Messages: {detail.message_count}")

    # Get messages
    messages = client.list_messages("session-id")
    for m in messages.data:
        print(f"[{m.role}] {m.content}")

    # Delete session
    client.delete_session("session-id")

Configuration

from hawk import HawkClient, RetryConfig

client = HawkClient(
    base_url="http://localhost:4590",  # Daemon URL
    api_key="sk-...",                  # Optional API key
    timeout=60.0,                      # Request timeout in seconds
    retry_config=RetryConfig(
        max_retries=5,
        initial_backoff=1.0,
        max_backoff=60.0,
    ),
)

Error Handling

from hawk import HawkClient, NotFoundError, RateLimitError, HawkAPIError

with HawkClient() as client:
    try:
        session = client.get_session("nonexistent")
    except NotFoundError as e:
        print(f"Not found: {e.message}")
    except RateLimitError as e:
        print(f"Rate limited, retry after {e.retry_after}s")
    except HawkAPIError as e:
        print(f"API error {e.status_code}: {e.message}")

Development

pip install -e ".[dev]"
pytest

License

MIT