diff --git a/src/hawk/__init__.py b/src/hawk/__init__.py index 9b451a3..645ac00 100644 --- a/src/hawk/__init__.py +++ b/src/hawk/__init__.py @@ -3,6 +3,26 @@ from ._version import __version__ from .agent import Agent, AgentConfig, AsyncAgent from .client import AsyncHawkClient, HawkClient +from .composio_cognee import ( + CogneeFeedbackEntry, + CogneeImproveResult, + CogneeQAEntry, + CogneeSkillRunEntry, + CogneeTraceEntry, + ComposioCredential, + ComposioTool, + ComposioToolResult, + execute_composio_tool, + improve, + list_composio_credentials, + recall_with_session, + remember_feedback, + remember_qa, + remember_skill_run, + remember_trace, + search_composio_tools, + sync_session_to_permanent, +) from .discovery import ( AgentCard, AgentResolver, @@ -89,6 +109,14 @@ "BenchmarkResults", "ChatRequest", "ChatResponse", + "CogneeFeedbackEntry", + "CogneeImproveResult", + "CogneeQAEntry", + "CogneeSkillRunEntry", + "CogneeTraceEntry", + "ComposioCredential", + "ComposioTool", + "ComposioToolResult", "CompositeResolver", "EvalResult", "EvalTask", @@ -141,9 +169,19 @@ "chat_with_tools_async", "configure_tracing", "detect_provider", + "execute_composio_tool", + "improve", "is_tracing_enabled", + "list_composio_credentials", + "recall_with_session", + "remember_feedback", + "remember_qa", + "remember_skill_run", + "remember_trace", "run_benchmark", "run_benchmark_async", + "search_composio_tools", + "sync_session_to_permanent", "tool", "trace", "trace_chat", diff --git a/src/hawk/composio_cognee.py b/src/hawk/composio_cognee.py new file mode 100644 index 0000000..5c05303 --- /dev/null +++ b/src/hawk/composio_cognee.py @@ -0,0 +1,439 @@ +"""Composio and cognee integration for the Hawk SDK. + +This module provides Python methods for: + - Searching composio tools + - Executing composio tools + - Listing composio credentials + - Storing and recalling cognee structured memory entries (QA, trace, feedback, skill run) + - Improving memories (re-process for quality) + - Syncing session memories to permanent graph +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from .client import DEFAULT_BASE_URL, DEFAULT_TIMEOUT +from .errors import parse_error +from .retry import DEFAULT_RETRY_CONFIG, RetryConfig, with_retry_sync + +# --- Composio Types --- + + +@dataclass +class ComposioTool: + name: str + description: str = "" + scope: str = "read" + auth_required: bool = False + params: dict[str, Any] = field(default_factory=dict) + tags: list[str] = field(default_factory=list) + category: str = "" + + +@dataclass +class ComposioToolResult: + success: bool = False + data: dict[str, Any] = field(default_factory=dict) + error: str = "" + + +@dataclass +class ComposioCredential: + id: str + service_name: str + type: str + scope: str = "" + expires_at: str | None = None + metadata: dict[str, str] = field(default_factory=dict) + + +# --- Cognee Entry Types --- + + +@dataclass +class CogneeQAEntry: + question: str + answer: str + project: str + context: str = "" + feedback_text: str = "" + feedback_score: int | None = None + used_graph_ids: list[str] = field(default_factory=list) + session_id: str = "" + source_agent: str = "" + + +@dataclass +class CogneeTraceEntry: + origin_function: str + project: str + session_id: str = "" + status: str = "success" + method_params: str = "" + method_return_value: str = "" + memory_query: str = "" + memory_context: str = "" + error_message: str = "" + source_agent: str = "" + + +@dataclass +class CogneeFeedbackEntry: + target_node_id: str + project: str + feedback_text: str = "" + score: int | None = None + session_id: str = "" + source_agent: str = "" + + +@dataclass +class CogneeSkillRunEntry: + skill_name: str + project: str + status: str = "success" + skill_version: str = "" + params: str = "" + result: str = "" + duration_ms: int = 0 + error: str = "" + session_id: str = "" + source_agent: str = "" + + +@dataclass +class CogneeImproveResult: + nodes_processed: int = 0 + nodes_improved: int = 0 + summaries_generated: int = 0 + embeddings_refreshed: int = 0 + duplicates_merged: int = 0 + errors: int = 0 + duration_ms: int = 0 + + +# --- Composio Methods --- + + +def search_composio_tools( + base_url: str = DEFAULT_BASE_URL, + query: str = "", + timeout: float = DEFAULT_TIMEOUT, + retry_config: RetryConfig | None = None, +) -> list[ComposioTool]: + """Search the composio tool catalog.""" + import httpx + + if retry_config is None: + retry_config = DEFAULT_RETRY_CONFIG + + def _do_search() -> list[ComposioTool]: + with httpx.Client(timeout=timeout) as client: + resp = client.post( + f"{base_url}/composio/search", + json={"query": query}, + ) + if resp.status_code != 200: + raise parse_error(resp) + data = resp.json() + return [ComposioTool(**t) for t in data.get("tools", [])] + + return with_retry_sync(_do_search, retry_config) + + +def execute_composio_tool( + base_url: str = DEFAULT_BASE_URL, + name: str = "", + params: dict[str, Any] | None = None, + timeout: float = DEFAULT_TIMEOUT, + retry_config: RetryConfig | None = None, +) -> ComposioToolResult: + """Execute a composio tool by name.""" + import httpx + + if retry_config is None: + retry_config = DEFAULT_RETRY_CONFIG + if params is None: + params = {} + + def _do_execute() -> ComposioToolResult: + with httpx.Client(timeout=timeout) as client: + resp = client.post( + f"{base_url}/composio/execute", + json={"name": name, "params": params}, + ) + if resp.status_code != 200: + raise parse_error(resp) + return ComposioToolResult(**resp.json()) + + return with_retry_sync(_do_execute, retry_config) + + +def list_composio_credentials( + base_url: str = DEFAULT_BASE_URL, + timeout: float = DEFAULT_TIMEOUT, + retry_config: RetryConfig | None = None, +) -> list[ComposioCredential]: + """List all composio credentials.""" + import httpx + + if retry_config is None: + retry_config = DEFAULT_RETRY_CONFIG + + def _do_list() -> list[ComposioCredential]: + with httpx.Client(timeout=timeout) as client: + resp = client.get(f"{base_url}/composio/credentials") + if resp.status_code != 200: + raise parse_error(resp) + return [ComposioCredential(**c) for c in resp.json()] + + return with_retry_sync(_do_list, retry_config) + + +# --- Cognee Memory Methods --- + + +def remember_qa( + base_url: str = DEFAULT_BASE_URL, + qa: CogneeQAEntry | None = None, + timeout: float = DEFAULT_TIMEOUT, + retry_config: RetryConfig | None = None, +) -> str: + """Store a Q&A entry as a structured memory node.""" + import httpx + + if retry_config is None: + retry_config = DEFAULT_RETRY_CONFIG + if qa is None: + raise ValueError("qa entry is required") + + def _do_remember() -> str: + with httpx.Client(timeout=timeout) as client: + resp = client.post( + f"{base_url}/cognee/qa", + json={ + "question": qa.question, + "answer": qa.answer, + "context": qa.context, + "feedback_text": qa.feedback_text, + "feedback_score": qa.feedback_score, + "used_graph_ids": qa.used_graph_ids, + "session_id": qa.session_id, + "project": qa.project, + "source_agent": qa.source_agent, + }, + ) + if resp.status_code != 200: + raise parse_error(resp) + return str(resp.json().get("id", "")) + + return with_retry_sync(_do_remember, retry_config) + + +def remember_trace( + base_url: str = DEFAULT_BASE_URL, + trace: CogneeTraceEntry | None = None, + timeout: float = DEFAULT_TIMEOUT, + retry_config: RetryConfig | None = None, +) -> str: + """Store a trace entry as a structured memory node.""" + import httpx + + if retry_config is None: + retry_config = DEFAULT_RETRY_CONFIG + if trace is None: + raise ValueError("trace entry is required") + + def _do_remember() -> str: + with httpx.Client(timeout=timeout) as client: + resp = client.post( + f"{base_url}/cognee/trace", + json={ + "origin_function": trace.origin_function, + "status": trace.status, + "method_params": trace.method_params, + "method_return_value": trace.method_return_value, + "memory_query": trace.memory_query, + "memory_context": trace.memory_context, + "error_message": trace.error_message, + "session_id": trace.session_id, + "project": trace.project, + "source_agent": trace.source_agent, + }, + ) + if resp.status_code != 200: + raise parse_error(resp) + return str(resp.json().get("id", "")) + + return with_retry_sync(_do_remember, retry_config) + + +def remember_feedback( + base_url: str = DEFAULT_BASE_URL, + feedback: CogneeFeedbackEntry | None = None, + timeout: float = DEFAULT_TIMEOUT, + retry_config: RetryConfig | None = None, +) -> str: + """Store feedback on an existing QA entry.""" + import httpx + + if retry_config is None: + retry_config = DEFAULT_RETRY_CONFIG + if feedback is None: + raise ValueError("feedback entry is required") + + def _do_remember() -> str: + with httpx.Client(timeout=timeout) as client: + resp = client.post( + f"{base_url}/cognee/feedback", + json={ + "target_node_id": feedback.target_node_id, + "feedback_text": feedback.feedback_text, + "score": feedback.score, + "project": feedback.project, + "session_id": feedback.session_id, + "source_agent": feedback.source_agent, + }, + ) + if resp.status_code != 200: + raise parse_error(resp) + return str(resp.json().get("id", "")) + + return with_retry_sync(_do_remember, retry_config) + + +def remember_skill_run( + base_url: str = DEFAULT_BASE_URL, + skill_run: CogneeSkillRunEntry | None = None, + timeout: float = DEFAULT_TIMEOUT, + retry_config: RetryConfig | None = None, +) -> str: + """Store a skill run entry as a structured memory node.""" + import httpx + + if retry_config is None: + retry_config = DEFAULT_RETRY_CONFIG + if skill_run is None: + raise ValueError("skill_run entry is required") + + def _do_remember() -> str: + with httpx.Client(timeout=timeout) as client: + resp = client.post( + f"{base_url}/cognee/skill_run", + json={ + "skill_name": skill_run.skill_name, + "skill_version": skill_run.skill_version, + "params": skill_run.params, + "result": skill_run.result, + "status": skill_run.status, + "duration_ms": skill_run.duration_ms, + "error": skill_run.error, + "project": skill_run.project, + "session_id": skill_run.session_id, + "source_agent": skill_run.source_agent, + }, + ) + if resp.status_code != 200: + raise parse_error(resp) + return str(resp.json().get("id", "")) + + return with_retry_sync(_do_remember, retry_config) + + +def improve( + base_url: str = DEFAULT_BASE_URL, + project: str = "", + min_confidence: float = 0.0, + min_access_count: int = 0, + consolidate_duplicates: bool = True, + regenerate_summaries: bool = True, + refresh_embeddings: bool = False, + limit: int = 1000, + timeout: float = DEFAULT_TIMEOUT, + retry_config: RetryConfig | None = None, +) -> CogneeImproveResult: + """Re-process memories to enhance their quality.""" + import httpx + + if retry_config is None: + retry_config = DEFAULT_RETRY_CONFIG + + def _do_improve() -> CogneeImproveResult: + with httpx.Client(timeout=timeout) as client: + resp = client.post( + f"{base_url}/cognee/improve", + json={ + "project": project, + "min_confidence": min_confidence, + "min_access_count": min_access_count, + "consolidate_duplicates": consolidate_duplicates, + "regenerate_summaries": regenerate_summaries, + "refresh_embeddings": refresh_embeddings, + "limit": limit, + }, + ) + if resp.status_code != 200: + raise parse_error(resp) + return CogneeImproveResult(**resp.json()) + + return with_retry_sync(_do_improve, retry_config) + + +def sync_session_to_permanent( + base_url: str = DEFAULT_BASE_URL, + session_id: str = "", + timeout: float = DEFAULT_TIMEOUT, + retry_config: RetryConfig | None = None, +) -> int: + """Sync session-scoped memories to the permanent graph.""" + import httpx + + if retry_config is None: + retry_config = DEFAULT_RETRY_CONFIG + + def _do_sync() -> int: + with httpx.Client(timeout=timeout) as client: + resp = client.post( + f"{base_url}/cognee/sync_session", + json={"session_id": session_id}, + ) + if resp.status_code != 200: + raise parse_error(resp) + return int(resp.json().get("promoted", 0)) + + return with_retry_sync(_do_sync, retry_config) + + +def recall_with_session( + base_url: str = DEFAULT_BASE_URL, + query: str = "", + session_id: str = "", + project: str = "", + limit: int = 10, + timeout: float = DEFAULT_TIMEOUT, + retry_config: RetryConfig | None = None, +) -> str: + """Perform session-aware recall.""" + import httpx + + if retry_config is None: + retry_config = DEFAULT_RETRY_CONFIG + + def _do_recall() -> str: + with httpx.Client(timeout=timeout) as client: + resp = client.post( + f"{base_url}/cognee/recall_session", + json={ + "query": query, + "session_id": session_id, + "project": project, + "limit": limit, + }, + ) + if resp.status_code != 200: + raise parse_error(resp) + return str(resp.json().get("context", "")) + + return with_retry_sync(_do_recall, retry_config)