undefined

undefined

from hermes.agent import Agent
from hermes.config import Config

**Key execution flow:**
1. `process_message()` → Prompt assembly
2. Model inference → Tool calls extraction
3. Tool dispatch via `ToolRegistry`
4. Result aggregation → Memory storage
5. Response generation

from hermes.tools.registry import tool_registry
from hermes.tools.base import ToolResult

@tool_registry.register(
    name="custom_analyzer",
    description="Analyze code patterns",
    category="analysis",
    parameters={
        "file_path": {
            "type": "string",
            "description": "Path to file to analyze"
        },
        "pattern": {
            "type": "string", 
            "description": "Pattern to search for"
        }
    }
)
async def custom_analyzer(file_path: str, pattern: str, **kwargs) -> ToolResult:
    """Custom code analysis tool."""
    try:
        with open(file_path, 'r') as f:
            content = f.read()
        
        matches = re.findall(pattern, content)
        
        return ToolResult(
            success=True,
            data={"matches": matches, "count": len(matches)},
            message=f"Found {len(matches)} matches"
        )
    except Exception as e:
        return ToolResult(
            success=False,
            error=str(e)
        )

from hermes.tools.toolsets import Toolset, toolset_registry

@toolset_registry.register("code_analysis")
class CodeAnalysisToolset(Toolset):
    """Custom toolset for code analysis."""
    
    def get_tools(self):
        return [
            "custom_analyzer",
            "list_functions",
            "complexity_check"
        ]
    
    def get_description(self):
        return "Tools for analyzing code structure and patterns"

from hermes.memory.manager import MemoryManager
from hermes.memory.store import MemoryStore
from hermes.memory.provider import MemoryProvider

**Session search with FTS5:**

```python
from hermes.memory.session_db import SessionDB

session_db = SessionDB(db_path="~/.hermes/sessions.db")

undefined

from hermes.compression.compressor import ContextCompressor

compressor = ContextCompressor(
    model_client=client,
    max_tokens=128000,
    preserve_recent=5  # Keep last 5 messages uncompressed
)

undefined

from hermes.skills.manager import SkillsManager

skills_manager = SkillsManager(
    skills_dir="~/.hermes/skills",
    config=config
)

undefined

undefined

undefined

from hermes.tools.browser import browser_navigate, browser_interact

undefined

from hermes.tools.code_exec import execute_code

result = await execute_code(
    code="""
import numpy as np
data = np.random.rand(100)
print(f"Mean: {data.mean()}")
""",
    language="python",
    timeout=30,
    allowed_imports=["numpy", "pandas", "matplotlib"]
)

**Communication modes:**

```python

undefined

from hermes.gateway.platform_registry import platform_registry
from hermes.gateway.base import PlatformAdapter, PlatformMessage

@platform_registry.register("custom_chat")
class CustomChatAdapter(PlatformAdapter):
    """Custom messaging platform integration."""
    
    platform_name = "custom_chat"
    
    async def initialize(self):
        """Connect to platform API."""
        self.client = CustomChatClient(
            api_key=self.config.get("api_key")
        )
        await self.client.connect()
    
    async def receive_messages(self):
        """Poll for new messages."""
        async for raw_msg in self.client.stream_messages():
            yield PlatformMessage(
                platform="custom_chat",
                channel_id=raw_msg.channel,
                user_id=raw_msg.author_id,
                username=raw_msg.author_name,
                content=raw_msg.text,
                message_id=raw_msg.id,
                timestamp=raw_msg.created_at
            )
    
    async def send_message(self, channel_id: str, content: str, **kwargs):
        """Send response to platform."""
        await self.client.send(
            channel=channel_id,
            text=content
        )
    
    def get_channel_prompt(self, channel_id: str) -> str:
        """Optional: platform-specific instructions."""
        return "Respond in a friendly, casual tone suitable for chat."

**Built-in platform adapters:**
- Discord, Slack, Telegram, IRC
- WeChat, QQ, DingTalk, WeCom (企业微信)
- WhatsApp, Signal, Matrix
- BlueBubbles (iMessage), SMS
- 腾讯元宝 (Tencent Yuanbao)

from hermes.plugins.base import Plugin, plugin_registry

@plugin_registry.register
class DashboardPlugin(Plugin):
    """Web dashboard for monitoring agent activity."""
    
    name = "dashboard"
    version = "1.0.0"
    
    async def initialize(self, agent):
        """Setup plugin."""
        self.agent = agent
        self.app = create_dashboard_app()
        
        # Register custom commands
        agent.register_command(
            name="/dashboard",
            handler=self.open_dashboard,
            description="Open web dashboard"
        )
        
        # Hook into tool execution
        agent.register_hook(
            "before_tool_call",
            self.log_tool_call
        )
    
    async def log_tool_call(self, tool_name, parameters):
        """Log tool executions to dashboard."""
        await self.app.broadcast_event({
            "type": "tool_call",
            "tool": tool_name,
            "params": parameters,
            "timestamp": time.time()
        })
    
    async def open_dashboard(self, args):
        """Handle /dashboard command."""
        url = await self.app.get_url()
        return f"Dashboard: {url}"

undefined

from hermes.mcp.client import MCPClient

undefined

from hermes.routing.smart_router import SmartRouter

router = SmartRouter(
    default_model="claude-3-7-sonnet",
    short_message_model="claude-3-5-haiku",
    short_message_threshold=100  # tokens
)

undefined

from hermes.optimization.cache import CacheStrategy

undefined

from hermes.security.approval import ApprovalSystem

undefined

from hermes.errors import HermesError, ToolExecutionError
from hermes.errors.classifier import ErrorClassifier

classifier = ErrorClassifier()

try:
    result = await tool_function(**params)
except Exception as e:
    # Structured error classification
    error_info = classifier.classify(e)
    
    if error_info.category == "rate_limit":
        # Automatic retry with backoff
        await asyncio.sleep(error_info.retry_after)
        result = await tool_function(**params)
    
    elif error_info.category == "auth_failure":
        # Try fallback credentials
        alt_creds = credential_pool.get_next()
        result = await tool_function(**params, creds=alt_creds)
    
    elif error_info.recoverable:
        # Switch to fallback model
        fallback_model = config.get_fallback_model()
        result = await fallback_model.complete(...)
    
    else:
        # Propagate with context
        raise HermesError(
            message=f"Unrecoverable error in {tool_name}",
            original_error=e,
            context=error_info.context
        )

undefined

undefined

credentials:
  anthropic:
    pool:
      - api_key: ${ANTHROPIC_KEY_1}
        rate_limit: 1000
      - api_key: ${ANTHROPIC_KEY_2}
        rate_limit: 500
    selection_strategy: round_robin  # or least_used, weighted, failover
  
  openai:
    pool:
      - api_key: ${OPENAI_KEY_MAIN}
        organization: ${OPENAI_ORG}
      - api_key: ${OPENAI_KEY_BACKUP}

gateway:
  enabled: true
  platforms:
    discord:
      enabled: true
      token: ${DISCORD_TOKEN}
      allowed_channels:
        - "1234567890"
      admin_users:
        - "user#1234"
      channel_prompts:
        "1234567890": "You are a helpful coding assistant."
    
    slack:
      enabled: true
      token: ${SLACK_TOKEN}
      signing_secret: ${SLACK_SIGNING_SECRET}
      socket_mode: true
    
    wechat:
      enabled: true
      auto_login: true
      contact_whitelist:
        - "friend_name"
  
  session_management:
    timeout: 3600  # seconds
    max_per_user: 5
    pii_redaction: true

undefined

undefined

/exit or /quit          - Exit session
/reset                  - Clear conversation
/dump                   - Export state
/models                 - List available models
/switch <model>         - Switch model
/profile <name>         - Switch profile
/tools                  - List active tools
/skills                 - List loaded skills
/reload-skills          - Reload skill library
/memory search <query>  - Search memories
/help                   - Show commands

from hermes.providers.base import ProviderTransport
from hermes.providers.registry import provider_registry

@provider_registry.register("custom_llm")
class CustomLLMTransport(ProviderTransport):
    """Custom LLM provider integration."""
    
    async def create_completion(self, messages, model, **kwargs):
        """Send completion request."""
        response = await self.http_client.post(
            f"{self.base_url}/v1/chat/completions",
            json={
                "model": model,
                "messages": self._format_messages(messages),
                "tools": self._format_tools(kwargs.get("tools", []))
            },
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        
        return self._parse_response(response)
    
    async def stream_completion(self, messages, model, **kwargs):
        """Stream completion chunks."""
        async with self.http_client.stream(
            "POST",
            f"{self.base_url}/v1/chat/completions",
            json={"model": model, "messages": messages, "stream": True}
        ) as stream:
            async for line in stream.aiter_lines():
                if line.startswith("data: "):
                    yield self._parse_chunk(line)
    
    def _format_tools(self, tools):
        """Convert Hermes tool schema to provider format."""
        return [
            {
                "name": tool["name"],
                "description": tool["description"],
                "parameters": tool["parameters"]
            }
            for tool in tools
        ]

from hermes.context.references import ContextReferenceParser

parser = ContextReferenceParser(
    sandbox_root="/workspace",
    max_file_size=100000  # bytes
)

undefined

from hermes.tools.parallel import ParallelExecutor

executor = ParallelExecutor(max_workers=5)

undefined

from hermes.voice.stt import STTProvider
from hermes.voice.tts import TTSProvider

undefined

undefined

**Problem:** Memories not being recalled.

```python

undefined

undefined

**Problem:** Parallel execution conflicts.

```python

undefined

undefined

**Problem:** PII leaking in logs.

```python

undefined

undefined

**Problem:** High API costs.

```python

undefined

undefined

undefined