Learn by building

from deepseek_harness import Harness

agent = Harness(model="deepseek-v4-flash")

# Register development tools
agent.register_tool("read_file", read_file)
agent.register_tool("write_file", write_file)
agent.register_tool("run_tests", pytest_runner)
agent.register_tool("git_diff", git_diff)

# Run the autonomous code fix cycle
result = agent.run(
    "Analyze src/ for bugs. Fix any issues found. "
    "Run the test suite. If tests fail, iterate until green. "
    "Output a summary report of all changes made.",
    max_iterations=25,
    retry_on_failure=True,
    memory_enabled=True,
)

print(result.summary)
# ✓ Found 3 bugs in src/auth.py
# ✓ Fixed null-pointer in validate_token()
# ✓ Fixed race condition in refresh_session()
# ✓ Fixed off-by-one in token expiry check
# ✓ All 42 tests passing
# ✓ Changes committed to branch fix/auth-bugs

from deepseek_harness import Harness

agent = Harness(model="deepseek-v4-flash")

# File system tools
agent.register_tool("list_files", list_directory)
agent.register_tool("read_file", read_file)
agent.register_tool("move_file", move_file)
agent.register_tool("write_json", write_json)

# Watch and process incoming documents
agent.run(
    "Watch ./inbox for new PDF files. "
    "Extract key information (date, amount, vendor) from each. "
    "Rename files to YYYY-MM-DD_vendor_amount.pdf format. "
    "Move processed files to ./archive and log to ./processed.json.",
    schedule="*/5 * * * *",  # every 5 minutes
    max_iterations=10,
)

from deepseek_harness import Harness

agent = Harness(model="deepseek-v4-flash")

# Register multiple MCP servers
agent.register_mcp_server("filesystem", "./workspace")
agent.register_mcp_server("search", "mcp://localhost:3000")
agent.register_mcp_server("database", "mcp://db.local:5432")

# Harness auto-discovers all available tools
print(agent.list_tools())
# [search_web, fetch_url, query_db, read_file, write_file, ...]

# Run a task that spans multiple servers
result = agent.run(
    "Search for recent papers on 'agent orchestration'. "
    "Fetch the top 3 results. Extract key findings. "
    "Store summaries in the database. "
    "Save full text to ./workspace/papers/.",
    max_iterations=30,
    sandbox=True,
)

from deepseek_harness import Harness, TaskDAG

agent = Harness(model="deepseek-v4-flash")

# Define a task dependency graph
dag = TaskDAG()
dag.add_task("fetch_data", fetch_from_api)
dag.add_task("clean_data", clean_dataset, depends_on=["fetch_data"])
dag.add_task("train_model", train_ml_model, depends_on=["clean_data"])
dag.add_task("generate_report", create_report, depends_on=["train_model"])
dag.add_task("notify_team", send_slack, depends_on=["generate_report"])

# Execute with checkpointing
result = agent.run_dag(
    dag,
    checkpoint_dir="./checkpoints",
    parallel_branches=True,
    resume_on_failure=True,
)

# If train_model fails at iteration 12,
# Harness resumes from the last checkpoint — no re-doing fetch_data.