# Copyright (c) 2025 Agentspan # Licensed under the MIT License. See LICENSE file in the project root for details. """Software Bug Assistant — agent_tool + mcp_tool for bug triage. Native SDK version of ADK example 33. Demonstrates: - agent_tool wrapping a search sub-agent - mcp_tool for live GitHub issue/PR lookup on conductor-oss/conductor - @tool for local ticket CRUD (in-memory store) Requirements: - Conductor server with AgentTool + MCP support - AGENTSPAN_SERVER_URL=http://localhost:6767/api in .env or environment - AGENTSPAN_LLM_MODEL=openai/gpt-4o-mini in .env or environment - GH_TOKEN in .env or environment """ import os from datetime import datetime from agentspan.agents import Agent, AgentRuntime, agent_tool, tool, mcp_tool from settings import settings # ── In-memory ticket store (mirrors real conductor-oss/conductor issues) ── _tickets: dict[str, dict] = { "COND-001": { "id": "COND-001", "title": "TaskStatusListener not invoked for system task lifecycle transitions", "status": "open", "priority": "high", "github_issue": 847, "description": "TaskStatusListener notifications are only fully wired for " "worker tasks (SIMPLE/custom). Both synchronous and asynchronous " "system tasks miss lifecycle transition callbacks.", "created": "2026-03-10", }, "COND-002": { "id": "COND-002", "title": "Support reasonForIncompletion in fail_task event handlers", "status": "open", "priority": "medium", "github_issue": 858, "description": "When an event handler uses action: fail_task, there is no way " "to set reasonForIncompletion. Need to support this field so " "failed tasks have meaningful error messages.", "created": "2026-03-13", }, "COND-003": { "id": "COND-003", "title": "Optimize /workflowDefs page: paginate latest-versions API", "status": "open", "priority": "medium", "github_issue": 781, "description": "The UI /workflowDefs page calls GET /metadata/workflow which " "returns all versions of all workflows. This causes slow page " "loads. Need pagination for the latest-versions endpoint.", "created": "2026-02-18", }, } _next_id = 4 # ── Function tools ──────────────────────────────────────────────── @tool def get_current_date() -> dict: """Get today's date. Returns: Dictionary with the current date. """ return {"date": datetime.now().strftime("%Y-%m-%d")} @tool def search_tickets(query: str) -> dict: """Search the internal bug ticket database for Conductor issues. Args: query: Search term to match against ticket titles and descriptions. Returns: Dictionary with matching tickets. """ query_lower = query.lower() matches = [ t for t in _tickets.values() if query_lower in t["title"].lower() or query_lower in t["description"].lower() ] return {"query": query, "count": len(matches), "tickets": matches} @tool def create_ticket(title: str, description: str, priority: str = "medium") -> dict: """Create a new bug ticket in the internal tracker. Args: title: Short title for the bug. description: Detailed description of the issue. priority: Priority level (low, medium, high, critical). Returns: Dictionary with the created ticket. """ global _next_id ticket_id = f"COND-{_next_id:03d}" _next_id += 1 ticket = { "id": ticket_id, "title": title, "status": "open", "priority": priority, "description": description, "created": datetime.now().strftime("%Y-%m-%d"), } _tickets[ticket_id] = ticket return {"created": True, "ticket": ticket} @tool def update_ticket(ticket_id: str, status: str = "", priority: str = "") -> dict: """Update an existing bug ticket's status or priority. Args: ticket_id: The ticket ID (e.g. COND-001). status: New status (open, in_progress, resolved, closed). Leave empty to skip. priority: New priority (low, medium, high, critical). Leave empty to skip. Returns: Dictionary with the updated ticket or error. """ ticket = _tickets.get(ticket_id.upper()) if not ticket: return {"error": f"Ticket {ticket_id} not found"} if status: ticket["status"] = status if priority: ticket["priority"] = priority return {"updated": True, "ticket": ticket} # ── Search sub-agent (wrapped as agent_tool) ────────────────────── @tool def search_web(query: str) -> dict: """Search the web for information about a Conductor bug or workflow issue. Args: query: The search query. Returns: Dictionary with search results. """ results = { "task status listener": { "source": "Conductor Docs", "answer": "TaskStatusListener is only wired for SIMPLE tasks. System " "tasks like HTTP, INLINE, SUB_WORKFLOW bypass the listener " "because they complete synchronously within the decider loop.", }, "do_while loop": { "source": "GitHub PR #820", "answer": "DO_WHILE tasks with 'items' now pass validation without " "loopCondition. Fixed in PR #820 — the validator was " "unconditionally requiring loopCondition for all DO_WHILE tasks.", }, "event handler fail": { "source": "GitHub Issue #858", "answer": "Event handlers with action: fail_task cannot set " "reasonForIncompletion. A proposed fix adds an optional " "'reason' field to the fail_task action configuration.", }, "workflow def pagination": { "source": "GitHub Issue #781", "answer": "The /metadata/workflow endpoint returns all versions of all " "workflows causing slow UI loads. A pagination API for " "latest-versions is proposed to fix this.", }, } query_lower = query.lower() for key, val in results.items(): if key in query_lower: return {"query": query, "found": True, **val} return {"query": query, "found": False, "summary": "No specific results found."} search_agent = Agent( name="search_agent_54", model=settings.llm_model, instructions=( "You are a technical search assistant specializing in Conductor " "(conductor-oss/conductor) workflow orchestration. Use the search_web " "tool to find relevant information about bugs, errors, and Conductor " "configuration issues. Provide concise, actionable answers." ), tools=[search_web], ) # ── GitHub MCP tools (live access to conductor-oss/conductor) ───── github_mcp_url = os.environ.get( "GITHUB_MCP_URL", "https://api.githubcopilot.com/mcp/" ) github_token = os.environ.get("GH_TOKEN", "") github = mcp_tool( server_url=github_mcp_url, name="github_mcp", description="GitHub tools for accessing the conductor-oss/conductor repository — " "search issues, list open pull requests, and get issue details", headers={"Authorization": f"Bearer {github_token}"}, tool_names=[ "search_repositories", "search_issues", "list_issues", "get_issue", "list_pull_requests", "get_pull_request", ], ) # ── Root agent ──────────────────────────────────────────────────── software_assistant = Agent( name="software_assistant_54", model=settings.llm_model, instructions=( "You are a software bug triage assistant for the Conductor workflow " "orchestration engine (https://github.com/conductor-oss/conductor).\n\n" "Your capabilities:\n" "1. Search and manage internal bug tickets (search_tickets, create_ticket, " "update_ticket)\n" "2. Research Conductor issues using the search_agent tool\n" "3. Look up real GitHub issues and PRs on conductor-oss/conductor using " "the GitHub MCP tools\n" "4. Cross-reference GitHub issues with internal tickets\n\n" "When triaging:\n" "- Use GitHub MCP tools to fetch the latest issues and PRs from " "conductor-oss/conductor\n" "- Cross-reference with internal tickets (search_tickets)\n" "- Research any unfamiliar issues with the search_agent\n" "- Create internal tickets for new issues not yet tracked\n" "- Suggest next steps, referencing GitHub issue/PR numbers" ), tools=[ get_current_date, agent_tool(search_agent), github, search_tickets, create_ticket, update_ticket, ], ) if __name__ == "__main__": with AgentRuntime() as runtime: result = runtime.run( software_assistant, "Review the latest open issues and PRs on conductor-oss/conductor. " "Check if any of them relate to our internal tickets. " "Pay attention to the DO_WHILE fix (PR #820) and the scheduler " "persistence PRs. Give me a triage summary.", ) result.print_result() # Production pattern: # 1. Deploy once during CI/CD: # runtime.deploy(software_assistant) # CLI alternative: # agentspan deploy --package examples.54_software_bug_assistant # # 2. In a separate long-lived worker process: # runtime.serve(software_assistant)