feat: Generic CRM API tool - AI can control entire system
- Create call_crm_api tool: single tool that can call ANY CRM API endpoint - Inject OpenAPI spec into system prompt so AI knows all available endpoints - Always include call_crm_api in agent tools (not just via tool_ids) - Extend get_current_user to support internal header-based auth (X-Internal-Call, X-Tenant-Id, X-User-Id) for AI tool API access - No more manual tool-per-endpoint registration needed
This commit is contained in:
Agent Zero
@@ -0,0 +1,194 @@
|
||||
"""Generic CRM API tool — gives AI direct access to all CRM REST endpoints.
|
||||
|
||||
Instead of registering dozens of individual tools, this single tool lets the AI
|
||||
call any CRM API endpoint. The OpenAPI spec is injected into the system prompt
|
||||
so the AI knows which endpoints exist and what parameters they accept.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.config import get_settings
|
||||
from app.core.db import get_session_factory
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Cache for OpenAPI spec
|
||||
_openapi_cache: dict[str, Any] | None = None
|
||||
|
||||
|
||||
def _get_base_url() -> str:
|
||||
"""Get the internal base URL for API calls."""
|
||||
import os
|
||||
port = os.environ.get("PORT", "8000")
|
||||
return f"http://127.0.0.1:{port}"
|
||||
|
||||
|
||||
async def get_openapi_spec() -> dict[str, Any]:
|
||||
"""Get the CRM OpenAPI spec, cached."""
|
||||
global _openapi_cache
|
||||
if _openapi_cache is not None:
|
||||
return _openapi_cache
|
||||
|
||||
try:
|
||||
base_url = _get_base_url()
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.get(f"{base_url}/openapi.json", timeout=10.0)
|
||||
if resp.status_code == 200:
|
||||
spec = resp.json()
|
||||
# Compact the spec: only keep paths, methods, summary, parameters
|
||||
compact: dict[str, Any] = {"paths": {}}
|
||||
for path, methods in spec.get("paths", {}).items():
|
||||
compact["paths"][path] = {}
|
||||
for method, details in methods.items():
|
||||
if method in ("get", "post", "patch", "put", "delete"):
|
||||
compact["paths"][path][method] = {
|
||||
"summary": details.get("summary", ""),
|
||||
"description": details.get("description", "")[:200],
|
||||
"parameters": [
|
||||
{
|
||||
"name": p.get("name", ""),
|
||||
"in": p.get("in", ""),
|
||||
"required": p.get("required", False),
|
||||
"schema": p.get("schema", {}),
|
||||
}
|
||||
for p in details.get("parameters", [])
|
||||
],
|
||||
"requestBody": details.get("requestBody", {}).get("content", {}),
|
||||
}
|
||||
_openapi_cache = compact
|
||||
return _openapi_cache
|
||||
except Exception as e:
|
||||
logger.warning("Failed to fetch OpenAPI spec: %s", e)
|
||||
|
||||
_openapi_cache = {"paths": {}}
|
||||
return _openapi_cache
|
||||
|
||||
|
||||
async def get_api_context_for_prompt() -> str:
|
||||
"""Get a compact text summary of available API endpoints for the system prompt."""
|
||||
spec = await get_openapi_spec()
|
||||
lines: list[str] = ["\n\n## Verfügbare CRM API Endpunkte\n"]
|
||||
lines.append("Du kannst jeden dieser Endpunkte mit dem Tool 'call_crm_api' aufrufen.")
|
||||
lines.append("Parameter: method (GET/POST/PATCH/DELETE), path (z.B. /api/v1/contacts), body (JSON für POST/PATCH).\n")
|
||||
|
||||
for path, methods in spec.get("paths", {}).items():
|
||||
for method, details in methods.items():
|
||||
summary = details.get("summary", "")
|
||||
lines.append(f"- {method.upper()} {path}: {summary}")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
async def call_crm_api_handler(arguments: dict[str, Any], context: dict[str, Any]) -> str:
|
||||
"""Execute an arbitrary CRM API call on behalf of the AI.
|
||||
|
||||
The AI provides method, path, and optional body. The handler makes an
|
||||
internal HTTP request with the user's session context (tenant_id, user_id)
|
||||
and returns the response.
|
||||
"""
|
||||
method = arguments.get("method", "GET").upper()
|
||||
path = arguments.get("path", "")
|
||||
body = arguments.get("body")
|
||||
|
||||
if not path:
|
||||
return json.dumps({"error": "path is required"})
|
||||
|
||||
# Ensure path starts with /
|
||||
if not path.startswith("/"):
|
||||
path = "/" + path
|
||||
|
||||
try:
|
||||
base_url = _get_base_url()
|
||||
|
||||
# Get user context for auth
|
||||
tenant_id = context.get("tenant_id", "")
|
||||
user_id = context.get("user_id", "")
|
||||
|
||||
# Create a DB session to resolve a valid session token for this user
|
||||
# We'll use internal service-level auth bypass
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"X-Internal-Call": "true",
|
||||
"X-Tenant-Id": str(tenant_id),
|
||||
"X-User-Id": str(user_id),
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
if method == "GET":
|
||||
resp = await client.get(f"{base_url}{path}", headers=headers, timeout=30.0)
|
||||
elif method == "POST":
|
||||
resp = await client.post(f"{base_url}{path}", headers=headers, json=body, timeout=30.0)
|
||||
elif method == "PATCH":
|
||||
resp = await client.patch(f"{base_url}{path}", headers=headers, json=body, timeout=30.0)
|
||||
elif method == "PUT":
|
||||
resp = await client.put(f"{base_url}{path}", headers=headers, json=body, timeout=30.0)
|
||||
elif method == "DELETE":
|
||||
resp = await client.delete(f"{base_url}{path}", headers=headers, timeout=30.0)
|
||||
else:
|
||||
return json.dumps({"error": f"Unsupported method: {method}"})
|
||||
|
||||
# Return response body (truncated if too large)
|
||||
try:
|
||||
resp_data = resp.json()
|
||||
resp_text = json.dumps(resp_data, default=str)
|
||||
if len(resp_text) > 8000:
|
||||
resp_text = resp_text[:8000] + "\n... [truncated]"
|
||||
return resp_text
|
||||
except Exception:
|
||||
return resp.text[:8000]
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("call_crm_api failed")
|
||||
return json.dumps({"error": str(e)})
|
||||
|
||||
|
||||
# Tool definition for the registry
|
||||
TOOL_NAME = "call_crm_api"
|
||||
TOOL_DESCRIPTION = (
|
||||
"Rufe einen beliebigen CRM API Endpunkt auf. Verwende GET zum Lesen, "
|
||||
"POST zum Erstellen, PATCH zum Aktualisieren, DELETE zum Löschen. "
|
||||
"Der path muss mit /api/v1/ beginnen (z.B. /api/v1/contacts). "
|
||||
"Für POST/PATCH kann ein body (JSON Objekt) mitgegeben werden."
|
||||
)
|
||||
TOOL_PARAMETERS = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"method": {
|
||||
"type": "string",
|
||||
"enum": ["GET", "POST", "PATCH", "PUT", "DELETE"],
|
||||
"default": "GET",
|
||||
"description": "HTTP Methode",
|
||||
},
|
||||
"path": {
|
||||
"type": "string",
|
||||
"description": "API Pfad, z.B. /api/v1/contacts oder /api/v1/contacts/{id}",
|
||||
},
|
||||
"body": {
|
||||
"type": "object",
|
||||
"description": "Request body für POST/PATCH (JSON Objekt)",
|
||||
},
|
||||
},
|
||||
"required": ["method", "path"],
|
||||
}
|
||||
|
||||
|
||||
def register_crm_api_tool(registry) -> None:
|
||||
"""Register the generic CRM API tool."""
|
||||
registry.register(
|
||||
name=TOOL_NAME,
|
||||
description=TOOL_DESCRIPTION,
|
||||
parameters=TOOL_PARAMETERS,
|
||||
handler=call_crm_api_handler,
|
||||
plugin_name="ai_assistant",
|
||||
required_permission=None, # Uses internal auth context
|
||||
category="system",
|
||||
)
|
||||
logger.info("CRM API tool registered — AI can now call any CRM endpoint")
|
||||
@@ -63,9 +63,18 @@ class AIAssistantPlugin(BasePlugin):
|
||||
await seed_defaults(db)
|
||||
|
||||
async def on_activate(self, db, service_container, event_bus) -> None:
|
||||
"""Activate plugin: register context tools and participant handler."""
|
||||
"""Activate plugin: register CRM API tool and participant handler."""
|
||||
await super().on_activate(db, service_container, event_bus)
|
||||
|
||||
# Register the generic CRM API tool — gives AI full system access
|
||||
try:
|
||||
from app.plugins.builtins.ai_assistant.crm_api_tool import register_crm_api_tool
|
||||
from app.plugins.builtins.ai_assistant.tool_registry import get_tool_registry
|
||||
register_crm_api_tool(get_tool_registry())
|
||||
logger.info("CRM API tool registered — AI has full system access")
|
||||
except Exception:
|
||||
logger.exception("Failed to register CRM API tool")
|
||||
|
||||
# Register as participant in the kommunikation system
|
||||
try:
|
||||
from app.plugins.builtins.ai_assistant.participant_handler import (
|
||||
|
||||
@@ -308,6 +308,13 @@ async def build_litellm_params(
|
||||
# Build messages with system prompt
|
||||
litellm_messages = []
|
||||
if system_prompt:
|
||||
# Inject CRM API context into system prompt
|
||||
try:
|
||||
from app.plugins.builtins.ai_assistant.crm_api_tool import get_api_context_for_prompt
|
||||
api_context = await get_api_context_for_prompt()
|
||||
system_prompt = system_prompt + api_context
|
||||
except Exception:
|
||||
logger.warning("Failed to inject API context into system prompt")
|
||||
litellm_messages.append({"role": "system", "content": system_prompt})
|
||||
litellm_messages.extend(messages)
|
||||
|
||||
@@ -427,9 +434,13 @@ async def stream_chat(
|
||||
messages.append({"role": "user", "content": full_message})
|
||||
await save_message(db, session.id, "user", user_message, tenant_id)
|
||||
|
||||
# Get agent tools
|
||||
# Get agent tools — always include call_crm_api for full system access
|
||||
registry = get_tool_registry()
|
||||
tools = registry.get_by_names(agent.tool_ids or [])
|
||||
# Ensure call_crm_api is always available
|
||||
crm_api_tool = registry.get("call_crm_api")
|
||||
if crm_api_tool and crm_api_tool not in tools:
|
||||
tools.append(crm_api_tool)
|
||||
tool_schemas = [t.to_openai_schema() for t in tools] if tools else None
|
||||
|
||||
# Build LLM params
|
||||
|
||||
Reference in New Issue
Block a user