Karta

from karta import Karta

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

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

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]

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)

def session(metadata=None, *, harness_args=None) -> Session

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

@app.on("message.completed")
async def log_response(event):
    print(event.session.id, event.message.text)

def policy(self, name: str, value) -> None

app.policy("max_messages_per_session", 50)