235 lines
8.4 KiB
Python
235 lines
8.4 KiB
Python
|
"""
|
||
|
|
Claude API agent with tool use.
|
||
|
|
All JARVIS capabilities are registered as tools that Claude can invoke.
|
||
|
|
"""
|
||
|
|
import os
|
||
|
|
import json
|
||
|
|
import anthropic
|
||
|
|
from capabilities import (
|
||
|
|
screen_vision,
|
||
|
|
calendar_access,
|
||
|
|
email_access,
|
||
|
|
notes_manager,
|
||
|
|
task_manager,
|
||
|
|
file_system,
|
||
|
|
terminal_control,
|
||
|
|
browser_control,
|
||
|
|
git_manager,
|
||
|
|
)
|
||
|
|
|
||
|
|
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
|
||
|
|
|
||
|
|
SYSTEM_PROMPT = """You are JARVIS (Just A Rather Very Intelligent System), a Windows-based AI assistant.
|
||
|
|
You have access to the following capabilities: screen vision, Google Calendar, Gmail, Google Tasks,
|
||
|
|
Google Keep notes, file system management, PowerShell terminal, Chrome browser automation, and Git.
|
||
|
|
Be concise, direct, and helpful. When using tools, always explain what you found."""
|
||
|
|
|
||
|
|
TOOLS = [
|
||
|
|
{
|
||
|
|
"name": "capture_screen",
|
||
|
|
"description": "Take a screenshot and identify active windows and open applications",
|
||
|
|
"input_schema": {"type": "object", "properties": {}, "required": []}
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "get_calendar_events",
|
||
|
|
"description": "Get today's events or upcoming events from Google Calendar",
|
||
|
|
"input_schema": {
|
||
|
|
"type": "object",
|
||
|
|
"properties": {
|
||
|
|
"days_ahead": {"type": "integer", "description": "Number of days ahead to fetch (default 0 = today only)", "default": 0}
|
||
|
|
},
|
||
|
|
"required": []
|
||
|
|
}
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "get_emails",
|
||
|
|
"description": "Get unread emails from Gmail or search by query",
|
||
|
|
"input_schema": {
|
||
|
|
"type": "object",
|
||
|
|
"properties": {
|
||
|
|
"query": {"type": "string", "description": "Gmail search query (e.g. 'from:boss@company.com' or 'subject:invoice')"},
|
||
|
|
"count": {"type": "integer", "description": "Max number of emails to return", "default": 10}
|
||
|
|
},
|
||
|
|
"required": []
|
||
|
|
}
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "get_tasks",
|
||
|
|
"description": "Get pending tasks from Google Tasks",
|
||
|
|
"input_schema": {"type": "object", "properties": {}, "required": []}
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "create_task",
|
||
|
|
"description": "Create a new task in Google Tasks",
|
||
|
|
"input_schema": {
|
||
|
|
"type": "object",
|
||
|
|
"properties": {
|
||
|
|
"title": {"type": "string", "description": "Task title"},
|
||
|
|
"notes": {"type": "string", "description": "Optional task notes"},
|
||
|
|
"due": {"type": "string", "description": "Optional due date in RFC 3339 format"}
|
||
|
|
},
|
||
|
|
"required": ["title"]
|
||
|
|
}
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "create_note",
|
||
|
|
"description": "Create a note in Google Keep",
|
||
|
|
"input_schema": {
|
||
|
|
"type": "object",
|
||
|
|
"properties": {
|
||
|
|
"title": {"type": "string", "description": "Note title"},
|
||
|
|
"content": {"type": "string", "description": "Note body text"}
|
||
|
|
},
|
||
|
|
"required": ["title", "content"]
|
||
|
|
}
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "run_powershell",
|
||
|
|
"description": "Execute a PowerShell command and return the output",
|
||
|
|
"input_schema": {
|
||
|
|
"type": "object",
|
||
|
|
"properties": {
|
||
|
|
"command": {"type": "string", "description": "PowerShell command to run"}
|
||
|
|
},
|
||
|
|
"required": ["command"]
|
||
|
|
}
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "get_git_status",
|
||
|
|
"description": "Get git status of a project directory",
|
||
|
|
"input_schema": {
|
||
|
|
"type": "object",
|
||
|
|
"properties": {
|
||
|
|
"path": {"type": "string", "description": "Absolute path to the git repository"}
|
||
|
|
},
|
||
|
|
"required": ["path"]
|
||
|
|
}
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "open_browser",
|
||
|
|
"description": "Open Chrome and navigate to a URL or perform a Google search",
|
||
|
|
"input_schema": {
|
||
|
|
"type": "object",
|
||
|
|
"properties": {
|
||
|
|
"url": {"type": "string", "description": "URL to navigate to"},
|
||
|
|
"search": {"type": "string", "description": "Search query (alternative to url)"}
|
||
|
|
},
|
||
|
|
"required": []
|
||
|
|
}
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "create_project_folder",
|
||
|
|
"description": "Create a new project folder with a README",
|
||
|
|
"input_schema": {
|
||
|
|
"type": "object",
|
||
|
|
"properties": {
|
||
|
|
"name": {"type": "string", "description": "Project name"},
|
||
|
|
"base_path": {"type": "string", "description": "Base directory path", "default": "C:/Projects"}
|
||
|
|
},
|
||
|
|
"required": ["name"]
|
||
|
|
}
|
||
|
|
}
|
||
|
|
]
|
||
|
|
|
||
|
|
def _dispatch_tool(name: str, inputs: dict) -> str:
|
||
|
|
"""Route a Claude tool_use call to the appropriate capability module."""
|
||
|
|
try:
|
||
|
|
if name == "capture_screen":
|
||
|
|
windows = screen_vision.list_open_windows()
|
||
|
|
active = screen_vision.get_active_window_title()
|
||
|
|
return json.dumps({"active_window": active, "open_windows": windows[:10]})
|
||
|
|
|
||
|
|
elif name == "get_calendar_events":
|
||
|
|
days = inputs.get("days_ahead", 0)
|
||
|
|
if days == 0:
|
||
|
|
events = calendar_access.get_todays_events()
|
||
|
|
else:
|
||
|
|
events = calendar_access.get_upcoming_events(days=days)
|
||
|
|
return json.dumps(events)
|
||
|
|
|
||
|
|
elif name == "get_emails":
|
||
|
|
query = inputs.get("query")
|
||
|
|
count = inputs.get("count", 10)
|
||
|
|
if query:
|
||
|
|
emails = email_access.search_emails(query)
|
||
|
|
else:
|
||
|
|
emails = email_access.get_unread_emails(count=count)
|
||
|
|
return json.dumps(emails)
|
||
|
|
|
||
|
|
elif name == "get_tasks":
|
||
|
|
return json.dumps(task_manager.get_pending_tasks())
|
||
|
|
|
||
|
|
elif name == "create_task":
|
||
|
|
result = task_manager.create_task(
|
||
|
|
title=inputs["title"],
|
||
|
|
notes=inputs.get("notes"),
|
||
|
|
due=inputs.get("due")
|
||
|
|
)
|
||
|
|
return json.dumps({"created": True, "id": result.get("id")})
|
||
|
|
|
||
|
|
elif name == "create_note":
|
||
|
|
note_id = notes_manager.create_note(inputs["title"], inputs["content"])
|
||
|
|
return json.dumps({"created": True, "id": note_id})
|
||
|
|
|
||
|
|
elif name == "run_powershell":
|
||
|
|
result = terminal_control.run_powershell(inputs["command"])
|
||
|
|
return json.dumps(result)
|
||
|
|
|
||
|
|
elif name == "get_git_status":
|
||
|
|
status = git_manager.get_git_status(inputs["path"])
|
||
|
|
return json.dumps({"status": status})
|
||
|
|
|
||
|
|
elif name == "open_browser":
|
||
|
|
if inputs.get("url"):
|
||
|
|
browser_control.navigate_to(inputs["url"])
|
||
|
|
return json.dumps({"opened": inputs["url"]})
|
||
|
|
elif inputs.get("search"):
|
||
|
|
browser_control.search_web(inputs["search"])
|
||
|
|
return json.dumps({"searching": inputs["search"]})
|
||
|
|
|
||
|
|
elif name == "create_project_folder":
|
||
|
|
path = file_system.create_project(
|
||
|
|
inputs["name"],
|
||
|
|
inputs.get("base_path", "C:/Projects")
|
||
|
|
)
|
||
|
|
return json.dumps({"created": path})
|
||
|
|
|
||
|
|
return json.dumps({"error": f"Unknown tool: {name}"})
|
||
|
|
except Exception as e:
|
||
|
|
return json.dumps({"error": str(e)})
|
||
|
|
|
||
|
|
async def get_response(user_text: str) -> str:
|
||
|
|
"""Send user text to Claude with tool use enabled. Returns final text response."""
|
||
|
|
messages = [{"role": "user", "content": user_text}]
|
||
|
|
|
||
|
|
while True:
|
||
|
|
response = client.messages.create(
|
||
|
|
model="claude-opus-4-5",
|
||
|
|
max_tokens=1024,
|
||
|
|
system=SYSTEM_PROMPT,
|
||
|
|
tools=TOOLS,
|
||
|
|
messages=messages
|
||
|
|
)
|
||
|
|
|
||
|
|
# If Claude wants to use a tool, dispatch and continue
|
||
|
|
if response.stop_reason == "tool_use":
|
||
|
|
tool_results = []
|
||
|
|
for block in response.content:
|
||
|
|
if block.type == "tool_use":
|
||
|
|
result = _dispatch_tool(block.name, block.input)
|
||
|
|
tool_results.append({
|
||
|
|
"type": "tool_result",
|
||
|
|
"tool_use_id": block.id,
|
||
|
|
"content": result
|
||
|
|
})
|
||
|
|
|
||
|
|
messages.append({"role": "assistant", "content": response.content})
|
||
|
|
messages.append({"role": "user", "content": tool_results})
|
||
|
|
|
||
|
|
else:
|
||
|
|
# Final text response
|
||
|
|
for block in response.content:
|
||
|
|
if hasattr(block, "text"):
|
||
|
|
return block.text
|
||
|
|
return "I couldn't generate a response."
|