Top AI Repos — open-source AI, indexed and scored
Top AI Repos tracks AI repositories on GitHub and answers two different questions about each one: is it moving right now, and would you bet a product on it.
Top AI Repos tracks AI repositories on GitHub and answers two different questions about each one: is it moving right now, and would you bet a product on it.
The production-ready agent harness framework for Python
| Date | Stars |
|---|---|
| 2026-07-24 | 293 |
| 2026-07-25 | 293 |
| 2026-07-28 | 293 |
| 2026-07-30 | 293 |
| 2026-08-06 | 332 |
Today
+39 stars today
This week
+39 stars this week
This month
— stars this month
Momentum
150.0
growth rate 13.31%/day
# Water
**The production-ready agent harness framework for Python.**
[](https://opensource.org/licenses/Apache-2.0)
[](https://pypi.org/project/water-ai/)
[](https://pypi.org/project/water-ai/)
## Overview
Water is an agent harness framework — it provides the infrastructure *around* your AI agents, not the agents themselves. Orchestration, resilience, observability, guardrails, approval gates, sandboxing, and deployment tooling so you can focus on what your agents actually do.
Works with any agent framework: LangChain, CrewAI, Agno, OpenAI, Anthropic, or your own custom agents.
## Installation
```bash
pip install water-ai
```
## Quick Start
```python
import asyncio
from water import Flow, create_task
from pydantic import BaseModel
class NumberInput(BaseModel):
value: int
class NumberOutput(BaseModel):
result: int
def add_five(params, context):
return {"result": params["input_data"]["value"] + 5}
task = create_task(
id="add",
description="Add five",
input_schema=NumberInput,
output_schema=NumberOutput,
execute=add_five,
)
flow = Flow(id="math", description="Math flow").then(task).register()
async def main():
result = await flow.run({"value": 10})
print(result) # {"result": 15}
asyncio.run(main())
```
## Flow Patterns
Water supports composable flow patterns that chain together with a fluent API:
```python
flow = Flow(id="pipeline", description="Example pipeline")
# Sequential — tasks run one after another
flow.then(task_a).then(task_b).then(task_c)
# Parallel — tasks run concurrently, results are merged
flow.parallel([task_a, task_b, task_c])
# Conditional branching — route to different tasks based on data
flow.branch([
(lambda data: data["type"] == "email", email_task),
(lambda data: data["type"] == "sms", sms_task),
])
# Loop — repeat a task while a condition holds
flow.loop(lambda data: data["retries"] < 3, retry_task, max_iterations=5)
# Map — run a task for each item in a list (parallel)
flow.map(process_task, over="items")
# DAG — define tasks with explicit dependencies
flow.dag(
[task_a, task_b, task_c],
dependencies={"task_c": ["task_a", "task_b"]},
)
# SubFlow composition — nest flows with input/output mapping
from water import SubFlow, compose_flows
sub = SubFlow(inner_flow, input_mapping={"text": "raw_input"}, output_mapping={"clean": "text"})
flow.then(sub.as_task())
# Compose multiple flows sequentially
pipeline = compose_flows(flow_a, flow_b, flow_c, id="full_pipeline")
# Try-catch-finally — structured error handling
flow.try_catch(
try_tasks=[risky_task, process_task],
catch_task=error_handler,
finally_task=cleanup_task,
)
# Conditional execution & fallbacks
flow.then(task, when=lambda data: data["enabled"])
flow.then(task, fallback=fallback_task)
```
## Agent Harness
Water provides infrastructure around your AI agents — not the agents themselves.
### LLM Tasks
Use any LLM provider through a unified interface:
```python
from water.agents import create_agent_task, OpenAIProvider, AnthropicProvider
agent = create_agent_task(
id="writer",
description="Write copy",
prompt_template="Write about: {topic}",
provider_instance=OpenAIProvider(model="gpt-4o"),
system_prompt="You are a copywriter.",
)
```
### Streaming LLM Agents
Stream responses token-by-token with real-time callbacks:
```python
from water.agents import create_streaming_agent_task, OpenAIStreamProvider
agent = create_streaming_agent_task(
id="stream_writer",
prompt_template="Write about: {topic}",
provider_instance=OpenAIStreamProvider(model="gpt-4o"),
on_chunk=lambda chunk: print(chunk.delta, end="", flush=True),
)
```
### Multi-Agent Orchestration
Coordinate multiple agents with shared context:
```python
from water.agents import create_agenExcerpt of 21,106 characters
Read on GitHubWould you bet a product on this? Bounded 0–100 and slow moving.
matched fp:b271456ad382d9f6, topic:agents, topic:multi-agent, topic:multi-agent-systems