Skip to content

API Reference

any_agent.AnyAgent

Bases: ABC

Base abstract class for all agent implementations.

This provides a unified interface for different agent frameworks.

Source code in src/any_agent/frameworks/any_agent.py
class AnyAgent(ABC):
    """Base abstract class for all agent implementations.

    This provides a unified interface for different agent frameworks.
    """

    @staticmethod
    def _get_agent_type_by_framework(
        framework_raw: AgentFramework | str,
    ) -> type[AnyAgent]:
        framework = AgentFramework.from_string(framework_raw)

        if framework is AgentFramework.SMOLAGENTS:
            from any_agent.frameworks.smolagents import SmolagentsAgent

            return SmolagentsAgent

        if framework is AgentFramework.LANGCHAIN:
            from any_agent.frameworks.langchain import LangchainAgent

            return LangchainAgent

        if framework is AgentFramework.OPENAI:
            from any_agent.frameworks.openai import OpenAIAgent

            return OpenAIAgent

        if framework is AgentFramework.LLAMA_INDEX:
            from any_agent.frameworks.llama_index import LlamaIndexAgent

            return LlamaIndexAgent

        if framework is AgentFramework.GOOGLE:
            from any_agent.frameworks.google import GoogleAgent

            return GoogleAgent

        if framework is AgentFramework.AGNO:
            from any_agent.frameworks.agno import AgnoAgent

            return AgnoAgent

        assert_never(framework)

    @classmethod
    def create(
        cls,
        agent_framework: AgentFramework | str,
        agent_config: AgentConfig,
        managed_agents: list[AgentConfig] | None = None,
    ) -> AnyAgent:
        agent_cls = cls._get_agent_type_by_framework(agent_framework)
        agent = agent_cls(agent_config, managed_agents=managed_agents)
        asyncio.get_event_loop().run_until_complete(agent._load_agent())
        return agent

    @abstractmethod
    async def _load_agent(self) -> None:
        """Load the agent instance."""

    def run(self, prompt: str) -> Any:
        """Run the agent with the given prompt."""
        return asyncio.get_event_loop().run_until_complete(self.run_async(prompt))

    @abstractmethod
    async def run_async(self, prompt: str) -> Any:
        """Run the agent asynchronously with the given prompt."""

    @property
    @abstractmethod
    def tools(self) -> list[Tool]:
        """Return the tools used by the agent.
        This property is read-only and cannot be modified.
        """

    def __init__(
        self,
        config: AgentConfig,
        managed_agents: list[AgentConfig] | None = None,
    ) -> None:
        msg = "Cannot instantiate the base class AnyAgent, please use the factory method 'AnyAgent.create'"
        raise NotImplementedError(msg)

    @property
    def agent(self) -> Any:
        """The underlying agent implementation from the framework.

        This property is intentionally restricted to maintain framework abstraction
        and prevent direct dependency on specific agent implementations.

        If you need functionality that relies on accessing the underlying agent:
        1. Consider if the functionality can be added to the AnyAgent interface
        2. Submit a GitHub issue describing your use case
        3. Contribute a PR implementing the needed functionality

        Raises:
            NotImplementedError: Always raised when this property is accessed

        """
        msg = "Cannot access the 'agent' property of AnyAgent, if you need to use functionality that relies on the underlying agent framework, please file a Github Issue or we welcome a PR to add the functionality to the AnyAgent class"
        raise NotImplementedError(msg)

agent property

The underlying agent implementation from the framework.

This property is intentionally restricted to maintain framework abstraction and prevent direct dependency on specific agent implementations.

If you need functionality that relies on accessing the underlying agent: 1. Consider if the functionality can be added to the AnyAgent interface 2. Submit a GitHub issue describing your use case 3. Contribute a PR implementing the needed functionality

Raises:

Type Description
NotImplementedError

Always raised when this property is accessed

tools abstractmethod property

Return the tools used by the agent. This property is read-only and cannot be modified.

run(prompt)

Run the agent with the given prompt.

Source code in src/any_agent/frameworks/any_agent.py
def run(self, prompt: str) -> Any:
    """Run the agent with the given prompt."""
    return asyncio.get_event_loop().run_until_complete(self.run_async(prompt))

run_async(prompt) abstractmethod async

Run the agent asynchronously with the given prompt.

Source code in src/any_agent/frameworks/any_agent.py
@abstractmethod
async def run_async(self, prompt: str) -> Any:
    """Run the agent asynchronously with the given prompt."""

any_agent.AgentFramework

Bases: str, Enum

Source code in src/any_agent/config.py
class AgentFramework(str, Enum):
    GOOGLE = auto()
    LANGCHAIN = auto()
    LLAMA_INDEX = auto()
    OPENAI = auto()
    AGNO = auto()
    SMOLAGENTS = auto()

    @classmethod
    def from_string(cls, value: str | Self) -> Self:
        if isinstance(value, cls):
            return value

        formatted_value = value.strip().upper()
        if formatted_value not in cls.__members__:
            error_message = (
                f"Unsupported agent framework: '{value}'. "
                f"Valid frameworks are: {list(cls.__members__.keys())}"
            )
            raise ValueError(error_message)

        return cls[formatted_value]

any_agent.AgentConfig

Bases: BaseModel

Source code in src/any_agent/config.py
class AgentConfig(BaseModel):
    model_config = ConfigDict(extra="forbid")
    model_id: str
    description: str = ""
    name: str = "any_agent"
    instructions: str | None = None
    tools: Sequence[Tool] = Field(default_factory=list)
    handoff: bool = False
    agent_type: str | None = None
    agent_args: MutableMapping[str, Any] | None = None
    model_type: str | None = None
    model_args: MutableMapping[str, Any] | None = None

any_agent.config.MCPStdioParams

Bases: BaseModel

Source code in src/any_agent/config.py
class MCPStdioParams(BaseModel):
    command: str
    args: Sequence[str]
    tools: Sequence[str] | None = None

any_agent.config.MCPSseParams

Bases: BaseModel

Source code in src/any_agent/config.py
class MCPSseParams(BaseModel):
    url: str
    headers: dict[str, str] | None = None
    tools: list[str] | None = None

any_agent.config.TracingConfig

Bases: BaseModel

Source code in src/any_agent/config.py
class TracingConfig(BaseModel):
    llm: str | None = "yellow"
    tool: str | None = "blue"
    agent: str | None = None
    chain: str | None = None

any_agent.tools

ask_user_verification(query)

Asks user to verify the given query.

Parameters:

Name Type Description Default
query str

The question that requires verification.

required
Source code in src/any_agent/tools/user_interaction.py
def ask_user_verification(query: str) -> str:
    """Asks user to verify the given `query`.

    Args:
        query: The question that requires verification.

    """
    return input(f"{query} => Type your answer here:")

search_web(query)

Performs a duckduckgo web search based on your query (think a Google search) then returns the top search results.

Parameters:

Name Type Description Default
query str

The search query to perform.

required

Returns:

Type Description
str

The top search results.

Source code in src/any_agent/tools/web_browsing.py
def search_web(query: str) -> str:
    """Performs a duckduckgo web search based on your query (think a Google search) then returns the top search results.

    Args:
        query (str): The search query to perform.

    Returns:
        The top search results.

    """
    ddgs = DDGS()
    results = ddgs.text(query, max_results=10)
    return "\n".join(
        f"[{result['title']}]({result['href']})\n{result['body']}" for result in results
    )

send_console_message(user, query)

Sends the specified user a message via console and returns their response.

Parameters:

Name Type Description Default
query str

The question to ask the user.

required
user str

The user to ask the question to.

required

Returns:

Name Type Description
str str

The user's response.

Source code in src/any_agent/tools/user_interaction.py
def send_console_message(user: str, query: str) -> str:
    """Sends the specified user a message via console and returns their response.

    Args:
        query: The question to ask the user.
        user: The user to ask the question to.

    Returns:
        str: The user's response.

    """
    return input(f"{query}\n{user}>>")

show_final_answer(answer)

Show the final answer to the user.

Parameters:

Name Type Description Default
answer str

The final answer.

required
Source code in src/any_agent/tools/user_interaction.py
def show_final_answer(answer: str) -> str:
    """Show the final answer to the user.

    Args:
        answer: The final answer.

    """
    logger.info(f"Final answer: {answer}")
    return answer

show_plan(plan)

Show the current plan to the user.

Parameters:

Name Type Description Default
plan str

The current plan.

required
Source code in src/any_agent/tools/user_interaction.py
def show_plan(plan: str) -> str:
    """Show the current plan to the user.

    Args:
        plan: The current plan.

    """
    logger.info(f"Current plan: {plan}")
    return plan

visit_webpage(url)

Visits a webpage at the given url and reads its content as a markdown string. Use this to browse webpages.

Parameters:

Name Type Description Default
url str

The url of the webpage to visit.

required
Source code in src/any_agent/tools/web_browsing.py
def visit_webpage(url: str) -> str:
    """Visits a webpage at the given url and reads its content as a markdown string. Use this to browse webpages.

    Args:
        url: The url of the webpage to visit.

    """
    try:
        response = requests.get(url)
        response.raise_for_status()

        markdown_content = markdownify(response.text).strip()

        markdown_content = re.sub(r"\n{2,}", "\n", markdown_content)

        return _truncate_content(markdown_content, 10000)
    except RequestException as e:
        return f"Error fetching the webpage: {e!s}"
    except Exception as e:
        return f"An unexpected error occurred: {e!s}"

any_agent.tracing

setup_tracing(agent_framework, output_dir='traces', tracing_config=None)

Setup tracing for agent_framework using openinference.instrumentation.

Parameters:

Name Type Description Default
agent_framework AgentFramework

The type of agent being used.

required
output_dir str

The directory where the traces will be stored. Defaults to "traces".

'traces'

Returns:

Name Type Description
str str

The name of the JSON file where traces will be stored.

Source code in src/any_agent/tracing.py
def setup_tracing(
    agent_framework: AgentFramework | str,
    output_dir: str | Path = "traces",
    tracing_config: TracingConfig | None = None,
) -> str:
    """Setup tracing for `agent_framework` using `openinference.instrumentation`.

    Args:
        agent_framework (AgentFramework): The type of agent being used.
        output_dir (str): The directory where the traces will be stored.
            Defaults to "traces".

    Returns:
        str: The name of the JSON file where traces will be stored.

    """
    agent_framework_ = AgentFramework.from_string(agent_framework)
    tracing_config = tracing_config or TracingConfig()

    tracer_provider, file_name = _get_tracer_provider(
        agent_framework_,
        output_dir,
        tracing_config,
    )

    instrumenter = get_instrumenter_by_framework(agent_framework_)
    instrumenter.instrument(tracer_provider=tracer_provider)

    return file_name