From e2cd43586196f6880620ca7cfaf246111470a538 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Sat, 25 Jul 2026 02:31:33 +0200 Subject: [PATCH] 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 --- app/deps.py | 30 ++- .../builtins/ai_assistant/crm_api_tool.py | 194 ++++++++++++++++++ app/plugins/builtins/ai_assistant/plugin.py | 11 +- app/plugins/builtins/ai_assistant/services.py | 13 +- 4 files changed, 245 insertions(+), 3 deletions(-) create mode 100644 app/plugins/builtins/ai_assistant/crm_api_tool.py diff --git a/app/deps.py b/app/deps.py index e6c6f63..97a3f25 100644 --- a/app/deps.py +++ b/app/deps.py @@ -24,12 +24,40 @@ async def get_current_user( db: AsyncSession = Depends(get_db), redis: aioredis.Redis = Depends(get_redis_dep), ) -> dict[str, Any]: - """Get the current authenticated user from session cookie. + """Get the current authenticated user from session cookie or internal headers. Returns session data dict with user_id, tenant_id, email, name, role, and resolved permissions from Redis cache. + + Supports internal calls via X-Internal-Call: true header with + X-Tenant-Id and X-User-Id headers (for AI tool API access). """ settings = get_settings() + + # Check for internal call (AI tool access) + if request.headers.get("X-Internal-Call") == "true": + tenant_id_str = request.headers.get("X-Tenant-Id", "") + user_id_str = request.headers.get("X-User-Id", "") + if tenant_id_str and user_id_str: + try: + tenant_id = uuid.UUID(tenant_id_str) + user_id = uuid.UUID(user_id_str) + await set_tenant_context(db, tenant_id) + + from app.core.permissions import get_cached_permissions + resolved = await get_cached_permissions(db, redis, user_id, tenant_id) + return { + "user_id": user_id_str, + "tenant_id": tenant_id_str, + "permissions": resolved.get("permissions", []), + "denied_permissions": resolved.get("denied", []), + "field_permissions": resolved.get("field_permissions", {}), + "is_system_admin": resolved.get("is_system_admin", False), + "is_active": True, + } + except (ValueError, Exception): + pass # Fall through to session cookie auth + session_id = request.cookies.get(settings.session_cookie_name) if not session_id: raise HTTPException( diff --git a/app/plugins/builtins/ai_assistant/crm_api_tool.py b/app/plugins/builtins/ai_assistant/crm_api_tool.py new file mode 100644 index 0000000..e1eb73c --- /dev/null +++ b/app/plugins/builtins/ai_assistant/crm_api_tool.py @@ -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") diff --git a/app/plugins/builtins/ai_assistant/plugin.py b/app/plugins/builtins/ai_assistant/plugin.py index 533f743..926ecdb 100644 --- a/app/plugins/builtins/ai_assistant/plugin.py +++ b/app/plugins/builtins/ai_assistant/plugin.py @@ -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 ( diff --git a/app/plugins/builtins/ai_assistant/services.py b/app/plugins/builtins/ai_assistant/services.py index 30f8068..1131553 100644 --- a/app/plugins/builtins/ai_assistant/services.py +++ b/app/plugins/builtins/ai_assistant/services.py @@ -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