> ## Documentation Index
> Fetch the complete documentation index at: https://docs.karta.sh/llms.txt
> Use this file to discover all available pages before exploring further.

# Karta

> The orchestrator - construct it, discover agents, send and stream turns, register hooks, set policies.

`Karta` is the core class. It detects your harness, discovers agents and skills,
holds session handles, and exposes `send`/`stream` in sync and async forms.

```python theme={null}
from karta import Karta
```

## Construct

```python theme={null}
Karta(
    agent_path: str | Path = ".",
    *,
    harness=None,                            # advanced: override harness auto-detection
    runnable=None,                           # advanced: supply a custom run function
)
```

```python theme={null}
app = Karta()                       # detect harness in the current directory
app = Karta("./my-app")             # point at a folder
app = Karta.from_path("./my-app")   # explicit classmethod
```

On construction, Karta auto-detects the harness, loads `karta.jsonc` if present,
and discovers agents and skills from the harness.

## Properties

<ResponseField name="agents" type="dict[str, Agent]">
  Discovered agents, keyed by name.
</ResponseField>

<ResponseField name="default_agent" type="Agent">
  The fallback agent used when none is specified.
</ResponseField>

<ResponseField name="skills" type="Mapping[str, Mapping]">
  Discovered skills and their metadata.
</ResponseField>

<ResponseField name="gateway" type="Gateway | None">
  The active [gateway](/api-reference/gateway), if configured.
</ResponseField>

<ResponseField name="profile" type="HumanAgent">
  The current user profile (from `karta.jsonc` or the OS user).
</ResponseField>

## Sending turns

All four take the same keyword arguments: `metadata`, `agent`, `session_id`,
`participant`, `thinking`.

```python theme={null}
async def send(text, *, metadata=None, agent=None, session_id=None,
               participant=None, thinking=False) -> Response
def      send_sync(...) -> Response
async def stream(...) -> AsyncIterator[StreamEvent]
def      stream_sync(...) -> Iterable[StreamEvent]
```

```python theme={null}
resp = app.send_sync("Audit invoice #789", agent="billing-specialist")
print(resp.text)

async for event in app.stream("Explain quicksort", thinking=True):
    if event.type == "text":
        print(event.data["part"]["text"], end="", flush=True)
```

A `Response` carries `text`, the turn `messages`, the `agent` that handled the
turn, and `usage` (`input_tokens`, `output_tokens`, `total_tokens`).

## Opening a session

```python theme={null}
def session(metadata=None, *, harness_args=None) -> Session
```

```python theme={null}
session = app.session(metadata={"customer_id": "abc123"})
session.send_sync("I need help with my order")
```

See the [Session reference](/sdks/python/sessions).

## Hooks

Register handlers for lifecycle events: `message.received`,
`message.completed`, `agent.handoff`, `session.created`.

```python theme={null}
@app.on("message.completed")
async def log_response(event):
    print(event.session.id, event.message.text)
```

## Policies

```python theme={null}
def policy(self, name: str, value) -> None
```

```python theme={null}
app.policy("max_messages_per_session", 50)
```

See [Hooks & policies](/sdks/python/hooks-and-policies) for the full set.