The Complete Production SOP: How to Deploy and Run CrewAI Custom Tools & Embeddings Memory in Autonomous AI Agents

from crewai.tools import tool
from pydantic import BaseModel, Field

class EnterpriseSearchInput(BaseModel):
    query: str = Field(..., description="The search phrase or keyword to retrieve documentation.")
    max_results: int = Field(default=5, ge=1, le=20, description="Maximum items to return.")

@tool("Enterprise Knowledge Base Search", args_schema=EnterpriseSearchInput)
def enterprise_search(query: str, max_results: int = 5) -> str:
    """Search internal documentation for architecture standards and operational runbooks."""
    try:
        # Simulated retrieval logic
        return f"Found {max_results} results for '{query}': [SOP 101, Arch RFC 42]"
    except Exception as exc:
        return f"Error querying knowledge base: {str(exc)}"

import os
from crewai import Agent, Crew, Process, Task
from src.tools.custom_search_tool import enterprise_search

def build_deterministic_crew() -> Crew:
    storage_dir = os.getenv("CREWAI_STORAGE_DIR", "./.crewai_storage")
    
    researcher = Agent(
        role="Senior Systems Architect",
        goal="Retrieve and summarize internal operational standards",
        backstory="You are an autonomous engineering auditor indexing technical documentation.",
        tools=[enterprise_search],
        verbose=True,
        memory=True
    )
    
    audit_task = Task(
        description="Execute knowledge search for 'production deployment' and extract requirements.",
        expected_output="A bulleted summary of all production deployment prerequisites.",
        agent=researcher
    )
    
    return Crew(
        agents=[researcher],
        tasks=[audit_task],
        process=Process.sequential,
        memory=True,
        verbose=True,
        embedder={
            "provider": "openai",
            "config": {
                "model": "text-embedding-3-small"
            }
        }
    )

Please create custom tool crewai for querying our PostgreSQL metrics database. Bind pydantic tool crewai schema with Host, Port, and Query fields, and enable memory in crewai Crew configuration.

import pytest
from src.tools.custom_search_tool import enterprise_search, EnterpriseSearchInput

def test_enterprise_search_valid():
    result = enterprise_search.run(query="Kubernetes ingress", max_results=3)
    assert "Found 3 results" in result

def test_enterprise_search_pydantic_validation():
    with pytest.raises(Exception):
        # max_results exceeds ge/le boundary
        enterprise_search.run(query="Security", max_results=50)