azcomputerguru/claudetools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync: auto-sync from GURU-BEAST-ROG at 2026-05-01 15:05:53

Browse Source 
Author: Mike Swanson
Machine: GURU-BEAST-ROG
Timestamp: 2026-05-01 15:05:53
This commit is contained in:
Mike Swanson
2026-05-01 15:05:53 -07:00
parent 2ec07ea409
commit e8e1874cbb
11 changed files with 429 additions and 559 deletions
Download Patch File Download Diff File Expand all files Collapse all files

214
projects/discord-bot/bot/claude/client.py
View File

@@ -1,149 +1,97 @@
"""Claude API client with streaming support for Discord."""
import asyncio
from datetime import datetime
from typing import Callable, Optional, Any
"""Claude Agent SDK wrapper for per-thread Discord conversations."""
from __future__ import annotations
import discord
from anthropic import AsyncAnthropic
from anthropic.types import MessageStreamEvent
import logging
from pathlib import Path
from typing import AsyncIterator, Awaitable, Callable, Optional
from claude_agent_sdk import (
AssistantMessage,
ClaudeAgentOptions,
ClaudeSDKClient,
ResultMessage,
TextBlock,
ToolUseBlock,
)
from bot.config import settings
from bot.claude.tools import TOOLS, SYSTEM_PROMPT_TEMPLATE
logger = logging.getLogger(__name__)
class ClaudeClient:
"""Wrapper around Anthropic SDK for Discord bot usage."""
def _load_system_prompt() -> str:
claude_md = settings.claudetools_root / ".claude" / "CLAUDE.md"
return claude_md.read_text(encoding="utf-8")
def __init__(self):
self.client = AsyncAnthropic(api_key=settings.anthropic_api_key)
self.model = settings.claude_model
def format_system_prompt(
self,
discord_user: discord.User,
channel_name: str,
thread_name: str,
user_role: str = "unknown"
) -> str:
"""Format system prompt with current context."""
return SYSTEM_PROMPT_TEMPLATE.format(
discord_username=discord_user.name,
discord_id=discord_user.id,
role=user_role,
channel_name=channel_name,
thread_name=thread_name,
datetime_utc=datetime.utcnow().isoformat()
class ThreadAgent:
"""One persistent Claude Code session bound to a Discord thread."""
def __init__(self, system_prompt: str, cwd: Path, model: str) -> None:
self._options = ClaudeAgentOptions(
system_prompt=system_prompt,
cwd=str(cwd),
model=model,
)
self._client: Optional[ClaudeSDKClient] = None
async def stream_response(
self,
messages: list[dict],
system_prompt: str,
tool_executor: Optional[Callable] = None,
progress_callback: Optional[Callable] = None
) -> tuple[str, list[dict]]:
"""
Stream a response from Claude, executing tools as needed.
async def start(self) -> None:
self._client = ClaudeSDKClient(options=self._options)
await self._client.connect()
Args:
messages: Conversation history
system_prompt: System prompt with context
tool_executor: Async function to execute tool calls
progress_callback: Async function to call with progress updates
async def stop(self) -> None:
if self._client is not None:
await self._client.disconnect()
self._client = None
Returns:
Tuple of (final_response_text, tool_results)
"""
final_text = ""
tool_results = []
async with self.client.messages.stream(
model=self.model,
max_tokens=4096,
system=system_prompt,
messages=messages,
tools=TOOLS
) as stream:
async for event in stream:
if event.type == "content_block_start":
if event.content_block.type == "tool_use":
# Tool call starting
tool_name = event.content_block.name
if progress_callback:
await progress_callback(f"🔧 Calling {tool_name}...")
elif event.type == "content_block_delta":
if hasattr(event.delta, "text"):
# Text content streaming
final_text += event.delta.text
if progress_callback and len(final_text) % 500 == 0:
# Send progress update every 500 chars
await progress_callback(final_text)
elif event.type == "message_stop":
# Check for tool uses
message = await stream.get_final_message()
for block in message.content:
if block.type == "tool_use":
# Execute tool
tool_name = block.name
tool_input = block.input
if tool_executor:
try:
if progress_callback:
await progress_callback(
f"⚙️ Executing {tool_name}..."
)
result = await tool_executor(tool_name, tool_input)
tool_results.append({
"name": tool_name,
"input": tool_input,
"result": result
})
if progress_callback:
await progress_callback(
f"✅ {tool_name} complete"
)
except Exception as e:
error_msg = f"Error in {tool_name}: {str(e)}"
tool_results.append({
"name": tool_name,
"input": tool_input,
"error": error_msg
})
if progress_callback:
await progress_callback(f"❌ {error_msg}")
elif block.type == "text":
final_text += block.text
return final_text, tool_results
async def simple_ask(
async def send(
self,
user_message: str,
conversation_history: list[dict],
system_prompt: str
on_text: Callable[[str], Awaitable[None]],
on_tool_use: Optional[Callable[[str], Awaitable[None]]] = None,
) -> str:
"""
Simple non-streaming request without tools.
Useful for summarization and simple queries.
"""
messages = conversation_history + [
{"role": "user", "content": user_message}
]
if self._client is None:
raise RuntimeError("ThreadAgent.send() called before start()")
response = await self.client.messages.create(
model=self.model,
max_tokens=4096,
system=system_prompt,
messages=messages
)
await self._client.query(user_message)
return response.content[0].text if response.content else ""
full_text = ""
async for message in self._client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
full_text += block.text
await on_text(block.text)
elif isinstance(block, ToolUseBlock) and on_tool_use is not None:
await on_tool_use(block.name)
elif isinstance(message, ResultMessage):
break
return full_text
class ClaudeAgentManager:
"""Owns one ThreadAgent per Discord thread id."""
def __init__(self) -> None:
self._system_prompt = _load_system_prompt()
self._cwd = settings.claudetools_root
self._model = settings.claude_model
self._agents: dict[int, ThreadAgent] = {}
async def get_or_create(self, thread_id: int) -> ThreadAgent:
agent = self._agents.get(thread_id)
if agent is None:
logger.info("[INFO] Starting new agent session for thread %d", thread_id)
agent = ThreadAgent(self._system_prompt, self._cwd, self._model)
await agent.start()
self._agents[thread_id] = agent
return agent
async def shutdown(self) -> None:
for thread_id, agent in list(self._agents.items()):
try:
await agent.stop()
except Exception as e:
logger.warning("[WARNING] Failed to stop agent %d: %s", thread_id, e)
self._agents.clear()

143
projects/discord-bot/bot/claude/tools.py
View File

@@ -1,142 +1 @@
"""Claude API tool definitions for ClaudeTools integration."""
TOOLS = [
{
"name": "query_claudetools_api",
"description": (
"Query the ClaudeTools MSP database. Use this for ALL data lookups including "
"clients, sessions, tasks, work items, billable time, infrastructure, "
"credentials, projects, and more. Returns JSON data from the API."
),
"input_schema": {
"type": "object",
"properties": {
"endpoint": {
"type": "string",
"description": (
"API endpoint path starting with /api/, e.g., '/api/clients', "
"'/api/sessions', '/api/tasks'"
)
},
"method": {
"type": "string",
"enum": ["GET", "POST", "PUT", "DELETE"],
"default": "GET",
"description": "HTTP method to use"
},
"params": {
"type": "object",
"description": (
"Query parameters as key-value pairs. Common params: "
"skip (offset), limit (page size), client_id, session_id, "
"status_filter, etc."
)
},
"body": {
"type": "object",
"description": "Request body for POST/PUT requests (JSON)"
}
},
"required": ["endpoint"]
}
},
{
"name": "run_breach_check",
"description": (
"Run a comprehensive 10-point M365 breach investigation on a single user account. "
"Checks: inbox rules, mailbox forwarding, OAuth consents, auth methods, "
"sign-ins (including foreign countries and legacy auth), directory audits, "
"risky user status, sent items, and deleted items. "
"Returns breach summary and artifact locations. "
"Requires tenant to be onboarded to remediation-tool."
),
"input_schema": {
"type": "object",
"properties": {
"tenant": {
"type": "string",
"description": (
"Tenant domain or GUID (e.g., 'cascadestucson.com' or "
"'4fcbb1f4-fbf9-4548-a93e-7d14a3c091e6')"
)
},
"upn": {
"type": "string",
"description": (
"User Principal Name - the user's email address "
"(e.g., 'john.trozzi@cascadestucson.com')"
)
}
},
"required": ["tenant", "upn"]
}
},
{
"name": "run_tenant_sweep",
"description": (
"Sweep an entire M365 tenant for security issues. "
"Checks: failed sign-ins from multiple foreign countries, "
"successful non-US sign-ins, B2B guest invitations, "
"consent/auth-method/role changes in directory audits, "
"and risky users (if IdentityRiskyUser consent granted). "
"Returns priority-sorted findings. "
"Requires tenant to be onboarded to remediation-tool."
),
"input_schema": {
"type": "object",
"properties": {
"tenant": {
"type": "string",
"description": (
"Tenant domain or GUID (e.g., 'dataforth.com' or "
"'dd4a82e8-85a3-44ac-8800-07945ab4d95f')"
)
}
},
"required": ["tenant"]
}
}
]
SYSTEM_PROMPT_TEMPLATE = """You are the ClaudeTools MSP Assistant for Arizona Computer Guru.
Available Tools:
1. query_claudetools_api - MSP database (clients, sessions, tasks, infrastructure, credentials)
2. run_breach_check - M365 user breach investigation (10-point audit)
3. run_tenant_sweep - M365 tenant-wide security sweep
Current Context:
- User: {discord_username} (Discord ID: {discord_id})
- Role: {role} (admin or tech)
- Channel: #{channel_name}
- Thread: {thread_name}
- DateTime: {datetime_utc}
Response Guidelines:
- Use Discord markdown: **bold**, `code`, ```language blocks```
- Keep responses under 2000 chars (Discord limit) - split into multiple messages if needed
- For structured data, use clear formatting or request embeds
- Ask before listing >5 items
- Security-conscious: NEVER expose credentials in responses
- Provide 1Password vault paths instead of actual secrets
Access Control:
- All team members: read-only queries, breach checks, tenant sweeps
- Mike/Howard only: remediation actions (require explicit confirmation)
- Dev/coding questions: refer to Mike or Howard
- NEVER execute destructive operations without explicit YES confirmation
Tool Usage:
- Use query_claudetools_api for ALL database lookups (don't make up data)
- Use run_breach_check for single-user M365 investigation
- Use run_tenant_sweep for tenant-wide M365 security analysis
- Chain tools when needed for complex multi-step queries
- Always cite which tool you used when presenting results
Remember:
- You're an MSP assistant - understand client/project/session/work item concepts
- Be concise but thorough
- If unsure, ask clarifying questions
- Guide users through multi-step processes
"""
"""Deprecated. Tools are provided by the Claude Agent SDK natively."""