Tip: How to Build a Custom Agent in Solace Agent Mesh

Solace Agent Mesh
1

Tip: How to Build a Custom Agent in Solace Agent Mesh

Solace Agent Mesh is built around intelligent agents that communicate through the A2A protocol to accomplish tasks. Solace Agent Mesh ships with an orchestrator agent, but the real power comes when you create your own domain-specific agents—whether for database queries, API integrations, document processing, or any custom capability.

Solace Agent Mesh provides a framework that handles all the hard parts—LLM orchestration, tool execution, session management, and agent discovery. You focus on one thing: writing the tools that define your agent’s capabilities.

Quick Start: A Working Agent in 5 Minutes

Here’s the minimum code to create a custom agent:

1. Create the tool function (src/my_agent/tools.py):

from typing import Any, Dict, Optional
from google.adk.tools import ToolContext
from solace_agent_mesh.agent.tools import ToolResult
from solace_ai_connector.common.log import log

async def greet_user(
    name: str,
    tool_context: Optional[ToolContext] = None,
    tool_config: Optional[Dict[str, Any]] = None
) -> ToolResult:
    """
    Greets a user with a personalized message.

    Args:
        name: The name of the person to greet.
    """
    log.info(f"[GreetUser] Greeting user: {name}")
    greeting_prefix = tool_config.get("greeting_prefix", "Hello") if tool_config else "Hello"
    return ToolResult.ok(f"{greeting_prefix}, {name}! Welcome to Agent Mesh!")

2. Configure the agent (configs/agents/my-agent.yaml):

apps:
  - name: my-agent
    app_module: solace_agent_mesh.agent.sac.app
    broker:
      <<: *broker_connection

    app_config:
      namespace: ${NAMESPACE}
      agent_name: "GreetingAgent"
      display_name: "Greeting Agent"
      supports_streaming: true
      
      model: *general_model
      model_provider:
        - "general"
      
      instruction: |
        You are a friendly greeting agent. Use the greet_user tool 
        to welcome users by name.
      
      tools:
        - tool_type: python
          component_module: "my_agent.tools"
          component_base_path: .
          function_name: "greet_user"
          tool_config:
            greeting_prefix: "Hello there"
      
      agent_card:
        description: "A friendly agent that greets users"
        defaultInputModes: ["text"]
        defaultOutputModes: ["text"]
        skills:
          - id: "greet_user"
            name: "Greet User"
            description: "Greets users with personalized messages"
      
      session_service: *default_session_service
      artifact_service: *default_artifact_service

That’s it. Run with sam run configs/agents/my-agent.yaml and your agent is live.

Table of Contents

Two Approaches

Add Agent (Project-specific): Use sam add agent my-agent to create an agent directly in your project. Best for one-off agents specific to your application.

Plugin (Reusable): Use sam plugin create my-agent-plugin --type agent to create a distributable package. Best for agents you want to share or reuse across projects.

See Agent or Plugin: Which to Use? for guidance.

Writing Tools

Tools are Python async functions that define what your agent can do. The LLM reads your docstring to decide when to use each tool.

Tool Function Requirements

from typing import Any, Dict, Optional
from google.adk.tools import ToolContext
from solace_agent_mesh.agent.tools import ToolResult
from solace_ai_connector.common.log import log

async def my_tool(
    param1: str,                                    # Required parameter
    param2: int = 10,                               # Optional with default
    tool_context: Optional[ToolContext] = None,     # Framework context
    tool_config: Optional[Dict[str, Any]] = None    # YAML configuration
) -> ToolResult:
    """
    Short description of what this tool does.

    Args:
        param1: Description for the LLM.
        param2: Another description.
    """
    log.info(f"[MyTool] Processing with param1={param1}")
    # Your logic here
    return ToolResult.ok("Success message", data={"key": "value"})

Key points:

  • Functions must be async def
  • The docstring becomes the tool’s description for the LLM
  • Type hints (str, int, bool) generate the parameter schema
  • Use ToolResult as the return type for consistent, structured responses
  • You can also return a Dict[str, Any] with a status field, but ToolResult is recommended

ToolResult

from solace_agent_mesh.agent.tools import ToolResult, DataObject, DataDisposition

# Simple success
return ToolResult.ok("Operation completed successfully")

# Success with data
return ToolResult.ok("Found 5 results", data={"count": 5, "items": [...]})

# Error
return ToolResult.error("Database connection failed")

# Success with file output
return ToolResult.ok(
    "Report generated",
    data_objects=[
        DataObject(
            name="report.csv",
            content=csv_content,
            mime_type="text/csv",
            disposition=DataDisposition.artifact
        )
    ]
)

Tool Patterns

Solace Agent Mesh supports three patterns for creating tools, from simple to advanced. See the Creating Python Tools guide for complete details.

Pattern 1: Function-Based (Simple)

Best for straightforward, self-contained tools:

async def calculate_sum(a: int, b: int) -> ToolResult:
    """Adds two numbers together."""
    return ToolResult.ok(f"The sum is {a + b}", data={"result": a + b})

tools:
  - tool_type: python
    component_module: "my_agent.tools"
    function_name: "calculate_sum"

Pattern 2: DynamicTool Class (Advanced)

Best for tools with complex logic or programmatic schemas:

from solace_agent_mesh.agent.tools.dynamic_tool import DynamicTool
from solace_agent_mesh.agent.tools import ToolResult
from google.genai import types as adk_types

class WeatherTool(DynamicTool):
    @property
    def tool_name(self) -> str:
        return "get_weather"

    @property
    def tool_description(self) -> str:
        return "Get current weather for a location."

    @property
    def parameters_schema(self) -> adk_types.Schema:
        return adk_types.Schema(
            type=adk_types.Type.OBJECT,
            properties={
                "location": adk_types.Schema(type=adk_types.Type.STRING, description="The city and state/country."),
                "units": adk_types.Schema(type=adk_types.Type.STRING, enum=["celsius", "fahrenheit"], nullable=True),
            },
            required=["location"],
        )

    async def _run_async_impl(self, args: dict, **kwargs) -> ToolResult:
        api_key = self.tool_config.get("api_key")
        if not api_key:
            return ToolResult.error("API key not configured")
        # ... call weather API ...
        return ToolResult.ok(f"The weather in {args['location']} is sunny.", data={"weather": "Sunny"})

Pattern 3: DynamicToolProvider (Factory)

Best for generating multiple related tools from a single config:

from typing import List
from solace_agent_mesh.agent.tools.dynamic_tool import DynamicTool, DynamicToolProvider
from solace_agent_mesh.agent.tools import ToolResult

class DatabaseToolProvider(DynamicToolProvider):
    def create_tools(self, tool_config=None) -> List[DynamicTool]:
        # Create tools from any decorated functions
        tools = self._create_tools_from_decorators(tool_config)
        # Add more complex tools programmatically
        if tool_config and tool_config.get("connection_string"):
            tools.append(DatabaseQueryTool(tool_config=tool_config))
        return tools

# NOTE: Decorator must be outside the class
@DatabaseToolProvider.register_tool
async def get_database_server_version(tool_config: dict, **kwargs) -> ToolResult:
    """Returns the version of the connected database server."""
    # ... implementation ...
    return ToolResult.ok("Database version retrieved.", data={"version": "PostgreSQL 15.3"})

Working with Artifacts

Use the Artifact type hint to automatically load files that users have uploaded or other tools have created:

from typing import Optional
from google.adk.tools import ToolContext
from solace_agent_mesh.agent.tools import Artifact, ToolResult

async def analyze_document(
    document: Artifact,
    include_word_count: bool = True,
    tool_context: Optional[ToolContext] = None,
) -> ToolResult:
    """
    Analyzes a document and returns statistics.

    Args:
        document: The document to analyze (pre-loaded by framework).
        include_word_count: Whether to include word count.
    """
    # Get content as text - handles bytes/str conversion automatically
    text = document.as_text()
    
    data = {
        "filename": document.filename,
        "version": document.version,
        "mime_type": document.mime_type,
        "character_count": len(text),
    }
    if include_word_count:
        data["word_count"] = len(text.split())
    
    return ToolResult.ok(f"Analyzed {document.filename}: {len(text)} characters.", data=data)

The LLM passes {"document": "report.txt"}, and your tool receives a full Artifact object with content, filename, version, mime_type, and metadata.

For multiple files, use List[Artifact]. For optional files, use Optional[Artifact].

See Working with Artifacts for more details.

Lifecycle Functions

Manage resources that persist across tool calls using lifecycle functions:

# src/my_agent/lifecycle.py
from typing import Any
from pydantic import BaseModel, Field
from solace_ai_connector.common.log import log

class MyAgentConfig(BaseModel):
    """Configuration model for agent initialization."""
    api_key: str = Field(..., description="API key for external service")
    cache_size: int = Field(default=100, description="Cache size limit")

def initialize_agent(host_component: Any, init_config: MyAgentConfig):
    """Called when the agent starts."""
    log_identifier = f"[{host_component.agent_name}:init]"
    log.info(f"{log_identifier} Starting agent initialization...")
    
    # Initialize shared resources
    host_component.set_agent_specific_state("api_client", create_client(init_config.api_key))
    host_component.set_agent_specific_state("request_count", 0)
    
    log.info(f"{log_identifier} Agent initialization completed")

def cleanup_agent(host_component: Any):
    """Called when the agent shuts down."""
    log_identifier = f"[{host_component.agent_name}:cleanup]"
    log.info(f"{log_identifier} Starting agent cleanup...")
    
    client = host_component.get_agent_specific_state("api_client")
    if client:
        client.close()
    
    request_count = host_component.get_agent_specific_state("request_count", 0)
    log.info(f"{log_identifier} Agent processed {request_count} requests during its lifetime")
    log.info(f"{log_identifier} Agent cleanup completed")

Reference them in your config:

app_config:
  agent_init_function:
    module: "my_agent.lifecycle"
    name: "initialize_agent"
    base_path: .
    config:
      api_key: ${MY_API_KEY}
      cache_size: 200
  
  agent_cleanup_function:
    module: "my_agent.lifecycle"
    name: "cleanup_agent"
    base_path: .

Agent Card

The agent card describes your agent’s capabilities to the mesh. Skills should match your configured tools:

agent_card:
  description: "An agent that processes documents and generates reports"
  defaultInputModes: ["text", "file"]
  defaultOutputModes: ["text", "file"]
  skills:
    - id: "analyze_document"
      name: "Analyze Document"
      description: "Analyzes uploaded documents and returns statistics"
    - id: "generate_report"
      name: "Generate Report"
      description: "Creates PDF reports from data"

The id should match the tool name (or tool_name if you’ve renamed it in config). The description helps the LLM (and other agents) decide when to use this skill.

Building and Distributing

Quick Debug Mode

During development, run directly from your source directory:

cd src
sam run ../config.yaml

Changes take effect immediately without rebuilding. This is useful for rapid iteration but should not be used for production.

Create a Plugin

For distributable agents:

sam plugin create my-agent-plugin --type agent

Build it (requires pip install build):

cd my-agent-plugin
sam plugin build

Add to a project:

sam plugin add my-custom-agent --plugin ./dist/my_agent_plugin-0.1.0.tar.gz

Or from PyPI:

sam plugin add my-custom-agent --plugin my-agent-plugin

Or from a Git repository:

sam plugin add my-custom-agent --plugin git+https://github.com/username/my-agent-plugin

See the Plugins documentation for more details.

Tips

  1. Start with one tool. Get a simple function working end-to-end before adding complexity.

  2. Write clear docstrings. The LLM uses them to decide when to call your tool. Be specific about inputs and outputs.

  3. Use ToolResult consistently. It provides structured responses that the LLM can parse reliably.

  4. Validate config with Pydantic. For DynamicTool classes, define a config_model to catch configuration errors at startup.

  5. Use lifecycle functions for shared resources. Database connections, API clients, and caches should be initialized once, not per-call.

  6. Use logging for debugging. Import from solace_ai_connector.common.log import log and add log statements to track tool execution.

  7. Check existing agents for patterns. The Weather Agent tutorial demonstrates a complete real-world implementation.

Learn More