ShipCrew Orchestration

from shipit_agent import Agent
from shipit_agent.deep.ship_crew import ShipCrew, ShipAgent, ShipTask

# Build two specialised agents
researcher_agent = Agent.with_builtins(llm=llm, prompt="You are a senior researcher.")
writer_agent = Agent.with_builtins(llm=llm, prompt="You are a technical writer.")

# Wrap them as ShipAgents with persona metadata
researcher = ShipAgent(
    name="researcher",
    agent=researcher_agent,
    role="Senior Researcher",
    goal="Find comprehensive, accurate information",
)
writer = ShipAgent(
    name="writer",
    agent=writer_agent,
    role="Technical Writer",
    goal="Produce clear, well-structured reports",
)

# Define tasks with dependencies
crew = ShipCrew(
    name="research-crew",
    coordinator_llm=llm,
    agents=[researcher, writer],
    tasks=[ShipTask(
            name="research",
            description="Research {topic} thoroughly",
            agent="researcher",
            output_key="findings",
        ),
        ShipTask(
            name="write",
            description="Write a report using these findings: {findings}",
            agent="writer",
            depends_on=["research"],
        ),],
)

result = crew.run(topic="AI agent architectures")
print(result.output)

# Diamond dependency pattern:
#   research
#   /      \
# analyze   summarize
#   \      /
#    report

tasks = [ShipTask(name="research", description="Research {topic}", agent="researcher"),
    ShipTask(
        name="analyze",
        description="Analyze findings: {research}",
        agent="analyst",
        depends_on=["research"],
    ),
    ShipTask(
        name="summarize",
        description="Summarize findings: {research}",
        agent="writer",
        depends_on=["research"],
    ),
    ShipTask(
        name="report",
        description="Combine analysis ({analyze}) and summary ({summarize})",
        agent="writer",
        depends_on=["analyze", "summarize"],
    ),]

result = crew.run(
    topic="microservices",
    language="Python",
    audience="senior engineers",
)

crew = ShipCrew(
    name="seq-crew",
    coordinator_llm=llm,
    agents=[researcher, writer],
    tasks=tasks,
    process="sequential",
)

crew = ShipCrew(
    name="fast-crew",
    coordinator_llm=llm,
    agents=[researcher, analyst, writer],
    tasks=tasks,
    process="parallel",
)
# "analyze" and "summarize" run concurrently since both only depend on "research"

crew = ShipCrew(
    name="smart-crew",
    coordinator_llm=llm,
    agents=[researcher, analyst, writer],
    tasks=tasks,
    process="hierarchical",
    max_rounds=10,
)

from shipit_agent.deep.ship_crew import ShipAgent

agent = ShipAgent(
    name="security-expert",
    agent=Agent.with_builtins(llm=llm, prompt="You are a security expert."),
    role="Security Auditor",
    goal="Identify and report security vulnerabilities",
    backstory="15 years of penetration testing experience.",
    capabilities=["code review", "OWASP analysis", "CVE lookup"],
)

# Direct execution (outside a crew)
result = agent.run("Audit the login module for SQL injection")

researcher = ShipAgent.from_registry("researcher", llm=llm)
auditor = ShipAgent.from_registry("security-auditor", llm=llm)

# With MCP servers
researcher = ShipAgent.from_registry(
    "researcher",
    llm=llm,
    mcps=[my_mcp_server],
)

# With overrides
custom = ShipAgent.from_registry(
    "code-reviewer",
    llm=llm,
    role="Python Code Reviewer",
    goal="Review Python code for PEP 8 compliance",
)

task = ShipTask(
    name="web-scrape",
    description="Scrape pricing data from {url}",
    agent="scraper",
    max_retries=3,
    timeout_seconds=120,
    context={"format": "JSON", "max_pages": "5"},
)

# To dict (for logging or persistence)
d = task.to_dict()

# From dict (for loading from JSON config)
restored = ShipTask.from_dict(d)

from shipit_agent.deep.ship_crew import create_ship_crew

crew = create_ship_crew(
    coordinator_llm=llm,
    agents=[{"name": "researcher", "agent": r_agent, "role": "Researcher"},
        writer_ship_agent,  # already a ShipAgent],
    tasks=[{"name": "research", "description": "Research {topic}", "agent": "researcher"},
        write_task,  # already a ShipTask],
    process="parallel",
    name="mixed-crew",
)

errors = crew.validate()
if errors:
    for err in errors:
        print(f"Config error: {err}")
else:
    result = crew.run(topic="AI safety")

from shipit_agent.deep.ship_crew.errors import (
    CyclicDependencyError,
    MissingAgentError,
    TaskTimeoutError,
    ShipCrewError,
)

try:
    result = crew.run(topic="AI agents")
except CyclicDependencyError as e:
    print(f"Fix your task graph: {e}")
except MissingAgentError as e:
    print(f"Register the agent first: {e}")
except TaskTimeoutError as e:
    print(f"Task took too long: {e}")
except ShipCrewError as e:
    print(f"Crew failed: {e}")

result = crew.run(topic="AI agents")

# Final output
print(result.output)

# Inspect individual task outputs
for key, value in result.task_results.items():
    print(f"\n--- {key} ---")
    print(value[:200])

# Execution metadata
print(f"Tasks: {result.total_tasks}")
print(f"Order: {result.execution_order}")
print(f"Failed: {result.failed_tasks}")
print(f"Time: {result.metadata['elapsed_seconds']}s")

# Serialize
d = result.to_dict()

for event in crew.stream(topic="AI agents"):
    print(f"[{event.type}] {event.message}")

from shipit_agent.costs import CostTracker, Budget

tracker = CostTracker(budget=Budget(max_dollars=10.00))
hooks = tracker.as_hooks()

# Attach hooks to each agent before wrapping in ShipAgent
r_agent = Agent.with_builtins(llm=llm, prompt="Researcher", hooks=hooks)
w_agent = Agent.with_builtins(llm=llm, prompt="Writer", hooks=hooks)

crew = ShipCrew(
    name="tracked-crew",
    coordinator_llm=llm,
    agents=[ShipAgent(name="researcher", agent=r_agent, role="Researcher"),
        ShipAgent(name="writer", agent=w_agent, role="Writer"),],
    tasks=[ShipTask(name="research", description="Research {topic}", agent="researcher"),
        ShipTask(name="write", description="Write about {research}", agent="writer",
                 depends_on=["research"]),],
)

result = crew.run(topic="quantum computing")

print(f"Total cost: ${tracker.total_cost:.4f}")
print(f"Calls: {len(tracker.breakdown())}")

from shipit_agent import Agent
from shipit_agent.deep.ship_crew import ShipCrew, ShipAgent, ShipTask, create_ship_crew
from shipit_agent.deep.ship_crew.errors import ShipCrewError
from shipit_agent.costs import CostTracker, Budget

# Cost tracking with $5 budget
tracker = CostTracker(
    budget=Budget(max_dollars=5.00, warn_at=0.70),
    on_cost_alert=lambda spent, limit: print(f"Cost alert: ${spent:.2f}/${limit:.2f}"),
)

# Build agents from registry
researcher = ShipAgent.from_registry("researcher", llm=llm)
writer = ShipAgent.from_registry("blog-writer", llm=llm)
reviewer = ShipAgent.from_registry("code-reviewer", llm=llm)

# Configure the crew
crew = ShipCrew(
    name="content-pipeline",
    coordinator_llm=llm,
    agents=[researcher, writer, reviewer],
    tasks=[ShipTask(
            name="research",
            description="Research {topic} with focus on practical examples",
            agent="Researcher",
            output_key="findings",
            max_retries=2,
        ),
        ShipTask(
            name="draft",
            description="Write a blog post about {topic} using: {findings}",
            agent="Blog Writer",
            depends_on=["research"],
            output_key="draft",
        ),
        ShipTask(
            name="review",
            description="Review this draft for technical accuracy: {draft}",
            agent="Code Reviewer",
            depends_on=["draft"],
            output_key="review",
            timeout_seconds=120,
        ),],
    process="sequential",
    verbose=True,
)

# Validate before running
errors = crew.validate()
if errors:
    raise ValueError(f"Crew config errors: {errors}")

# Execute with streaming
try:
    for event in crew.stream(topic="building AI agents with Python"):
        print(f"[{event.type}] {event.message}")
except ShipCrewError as e:
    print(f"Crew failed: {e}")

print(f"\nTotal cost: ${tracker.total_cost:.4f}")

from shipit_agent import Agent
from shipit_agent.deep.ship_crew import ShipCrew, ShipAgent, ShipTask

# Plain agents — no built-in tools, just LLM + prompt
researcher = ShipAgent(
    name="researcher",
    agent=Agent(llm=llm, prompt="You research topics thoroughly."),
    role="Researcher",
)

writer = ShipAgent(
    name="writer",
    agent=Agent(llm=llm, prompt="You write clear, concise summaries."),
    role="Writer",
)

crew = ShipCrew(
    name="simple-crew",
    coordinator_llm=llm,
    agents=[researcher, writer],
    tasks=[ShipTask(name="research", description="Research {topic}", agent="researcher", output_key="findings"),
        ShipTask(name="write", description="Summarize: {findings}", agent="writer", depends_on=["research"]),],
)

result = crew.run(topic="serverless computing trends")
print(result.output)

from shipit_agent.deep import DeepAgent

# DeepAgent as a crew member — gets planning, verification, and reflection
deep_researcher = DeepAgent.with_builtins(
    llm=llm,
    prompt="You are a thorough researcher with deep analysis capabilities.",
    verify=True,
)

crew = ShipCrew(
    name="deep-crew",
    coordinator_llm=llm,
    agents=[ShipAgent(name="deep-researcher", agent=deep_researcher.agent, role="Deep Researcher"),
        writer,],
    tasks=[ShipTask(name="research", description="Deep research on {topic}", agent="deep-researcher", output_key="findings"),
        ShipTask(name="write", description="Write report from: {findings}", agent="writer", depends_on=["research"]),],
)

result = crew.run(topic="AI safety in autonomous agents")

crew = ShipCrew(
    name="stream-detail",
    coordinator_llm=llm,
    agents=[researcher, writer],
    tasks=[ShipTask(name="research", description="Research Python async patterns", agent="researcher", output_key="findings"),
        ShipTask(name="write", description="Summarize: {findings}", agent="writer", depends_on=["research"]),],
)

for event in crew.stream():
    if event.type == "run_started":
        print(f"🚀 Crew started | mode: {event.payload.get('process')} | tasks: {event.payload.get('tasks')}")
    elif event.type == "tool_called":
        print(f"⚙️  Task '{event.payload.get('task')}' started → agent: {event.payload.get('agent')}")
    elif event.type == "tool_completed":
        print(f"✅ Task '{event.payload.get('task')}' done | output_key: {event.payload.get('output_key')}")
        print(f"   Preview: {event.payload.get('output', '')[:150]}...")
    elif event.type == "tool_failed":
        print(f"❌ Task '{event.payload.get('task')}' FAILED: {event.payload.get('error')}")
    elif event.type == "run_completed":
        print(f"🏁 Crew completed in {event.payload.get('elapsed_seconds')}s")
        print(f"   Order: {event.payload.get('execution_order')}")
        print(f"   Failed: {event.payload.get('failed_tasks')}")
        print(f"   Output: {event.payload.get('output', '')[:200]}")