Skip to main content
Version: 1.0.5

Fetch.ai Knowledge Assistant with MeTTa

Overview​

This guide shows how to integrate SingularityNET's MeTTa (Meta Type Talk) knowledge graph with Fetch.ai's uAgents framework. The sample is a toy demo: it looks up illustrative Fetch.ai / uAgents / Agentverse / ASI:One facts in MeTTa, then uses ASI:One to classify intent and humanize the reply. The seed graph is a snapshot (last reviewed 2026-08-22), not live product docs — prefer the Innovation Lab docs and ASI:One models for authoritative guidance.

Same MeTTa + uAgent skeleton as Medical Agent with MeTTa and Financial Advisor Agent with MeTTa.

Tested combo: Python 3.10–3.12, uagents>=0.25.5 (needs uagents-core 0.4.x), hyperon>=0.2.6. Chat Protocol samples need this runtime; Python 3.8 is not supported.

What is MeTTa?​

MeTTa (Meta Type Talk) is SingularityNET's multi-paradigm language for declarative and functional computations over knowledge (meta)graphs. Official docs: MeTTa language and Hyperon. It provides:

  • Structured Knowledge Representation: Organize platform/domain knowledge in queryable relations
  • Symbolic Reasoning: Perform logic-like pattern matching over graph atoms
  • Knowledge Graph Operations: Build, query, and evolve knowledge state
  • Space-based Architecture: Store knowledge as atoms in a logical space

Installation & Setup​

Prerequisites​

Before you begin, ensure you have:

  • Python 3.10+ (3.10–3.12 recommended for uagents 0.25.x). On Windows use py -3.10; on WSL/macOS/Linux use python3.
  • pip package manager
  • An ASI:One API key from the ASI:One API keys dashboard (not only the asi1.ai homepage)

Create a project folder and a virtual environment (do not install into system Python):

# macOS / Linux / WSL
python3 -m venv .venv
source .venv/bin/activate

# Windows PowerShell
py -3.10 -m venv .venv
.\.venv\Scripts\Activate.ps1

Create a .env file (never commit real secrets):

ASI_ONE_API_KEY=your_key_here
AGENT_SEED=change-me-to-a-unique-local-seed
# Optional: set LEARN=1 only if you want the demo to persist LLM guesses into the graph
# LEARN=1

Installation Options​

Create a requirements.txt file with one package per line:

openai>=1.0.0
hyperon>=0.2.6
uagents>=0.25.5
uagents-core>=0.4.9
python-dotenv>=1.0.0

uagents 0.25.x is tested with uagents-core 0.4.x (Chat Protocol). Keep uagents>=0.25.5 and uagents-core>=0.4.9 unless you intentionally upgrade the whole stack.

Install all dependencies with one command:

python3 -m pip install -r requirements.txt

On Windows: py -3.10 -m pip install -r requirements.txt.

Option 2: Verify Hyperon First​

Use this only to confirm Hyperon/MeTTa installs on your machine. You still need Option 1 (requirements.txt) for uagents, openai, and python-dotenv.

python3 -m pip install hyperon
python3 -c "from hyperon import MeTTa; print('Hyperon installed successfully!')"

Windows Installation Guide​

Hyperon on native Windows is often painful. WSL (Ubuntu) is recommended. If you stay on native Windows and hit build errors, see this video: Hyperon Installation on Windows.

Written WSL path:

  1. Install WSL and Ubuntu.
  2. Inside WSL: install Python 3.10+, create the venv above, then pip install -r requirements.txt.
  3. Run python3 agent.py from the project folder shown below.

Project layout​

Imports in agent.py use the metta package. Create this tree (a flat folder of four .py files will raise ImportError):

project/
  agent.py
  metta/
    __init__.py
    knowledge.py
    generalrag.py
    utils.py
  .env
  requirements.txt

Create empty metta/__init__.py. Run from project/:

python3 agent.py

Windows: py -3.10 agent.py.

This page is the Fetch.ai-domain sample of the shared MeTTa + uAgent skeleton. Copy the files below into that tree.

Architecture Overview​

Fetch.ai Knowledge Assistant — sequence-style workflow (yellow / green / white)

The code pipeline (ASI:One chat does not classify intent for you):

flowchart LR
  user[User or ASI:One chat]
  handler[Chat Protocol handler]
  llm[Agent LLM: intent plus keyword]
  lookup[MeTTa knowledge lookup]
  humanize[Humanize plus disclaimer]
  user --> handler --> llm --> lookup --> humanize --> user

Alt text: User or ASI:One sends chat text to the Chat Protocol handler. The agent LLM classifies intent and a keyword, MeTTa looks up the toy graph, then the agent humanizes the answer and sends a disclaimer-prefixed reply.

Architecture pipeline: User / ASI:One Chat → Chat Protocol handler → agent LLM classifies intent + keyword → MeTTa knowledge lookup (not vector RAG) → humanized reply with disclaimer → User.

Core Integration Concepts​

1. MeTTa Knowledge Graph Structure​

MeTTa organizes knowledge as atoms in logical spaces. Use one convention: capability / deployment / adapter keys as S(...); free-text solutions, considerations, and FAQ answers as ValueAtom. Multi-word names use underscores (hosted_agents, create_uagent), never raw spaces or parentheses inside query interpolation.

from hyperon import MeTTa, E, S, ValueAtom

metta = MeTTa()

metta.space().add_atom(E(S("capability"), S("uAgent"), S("message_handling")))
metta.space().add_atom(
    E(S("solution"), S("create_uagent"), ValueAtom("pip install uagents; define Agent; add handlers; run"))
)
metta.space().add_atom(E(S("faq"), S("hi"), ValueAtom("Hello! How can I help with uAgents today?")))

Key MeTTa Elements:

  • E (Expression): Creates logical expressions
  • S (Symbol): Represents symbolic atoms
  • ValueAtom: Stores string values (FAQ text, solution steps, considerations)
  • Space: Container where atoms are stored and queried

2. Pattern Matching and Querying​

# Query syntax: !(match &self (relation subject $variable) $variable)
query_str = '!(match &self (capability uAgent $feature) $feature)'
results = metta.run(query_str)
# Results include message_handling for the toy graph

Query Components:

  • &self: References the current space
  • $variable: Pattern matching variables that capture results
  • !(match ...): Query syntax for pattern matching

Never interpolate unsanitized user/LLM text into MeTTa. Only simple [a-z0-9_.:-]+ symbols are allowed (ASI:One model IDs use hyphens; ASI:One keeps a colon).

3. uAgent Chat Protocol Integration​

The following is an excerpt. Full Protocol construction is in agent.py. process_query returns a dict; send the humanized_answer string (plus disclaimer), not the dict.

from uagents_core.contrib.protocols.chat import (
    ChatMessage,
    ChatAcknowledgement,
    TextContent,
    chat_protocol_spec,
)

@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
    response = process_query(user_query, rag, llm)
    answer = response.get("humanized_answer", "I could not process that query.")
    await ctx.send(sender, create_text_chat(answer))

ctx.send delivers a message and pairs with acknowledgements (ChatAcknowledgement) in the Chat Protocol. Prefer ctx.send_and_receive when you need a synchronous request/response with a timeout.

mailbox=True on Agent(...) is the Agentverse mailbox flag (an inbox so a local agent stays reachable). Do not import mailbox — that is Python's stdlib email-mailbox module and is unused here. Older docs sometimes say Mailroom; the current term is Mailbox.

publish_agent_details=True publishes the agent's profile/details to Agentverse when the mailbox connects. Use it for discoverable demos; turn it off if you do not want the profile updated automatically.

4. Knowledge lookup (not vector RAG)​

GeneralRAG in this sample is a MeTTa retriever: pattern-match on the toy graph, then an LLM humanizes the result. It does not use embeddings or document RAG. Dynamic graph writes from LLM guesses are off unless LEARN=1 is set (unsafe for a product-knowledge assistant — invented APIs would be treated as graph truth).

Core Components​

  1. agent.py: Main uAgent with Chat Protocol
  2. metta/knowledge.py: Toy MeTTa graph (illustrative Fetch.ai snapshot — not live docs)
  3. metta/generalrag.py: MeTTa lookup helpers
  4. metta/utils.py: Intent classification and query processing

Implementation Guide​

Step 1: Define Your Knowledge Domain​

Create metta/knowledge.py. Keys are snake_case and must match what the classifier returns. Models match current ASI:One IDs: asi1, asi1-mini, asi1-ultra (models docs).

from hyperon import MeTTa, E, S, ValueAtom

# Toy snapshot last reviewed: 2026-08-22. Prefer Innovation Lab + ASI:One docs for live truth.


def initialize_knowledge_graph(metta: MeTTa):
    """Toy Fetch.ai / uAgents graph for the tutorial. Not official product docs."""
    # Agent types → capabilities (Symbol objects)
    metta.space().add_atom(E(S("capability"), S("uAgent"), S("microservice")))
    metta.space().add_atom(E(S("capability"), S("uAgent"), S("message_handling")))
    metta.space().add_atom(E(S("capability"), S("uAgent"), S("event_processing")))
    metta.space().add_atom(E(S("capability"), S("uAgent"), S("REST_endpoints")))
    metta.space().add_atom(E(S("capability"), S("ASI:One"), S("LLM_processing")))
    metta.space().add_atom(E(S("capability"), S("ASI:One"), S("agent_discovery")))
    metta.space().add_atom(E(S("capability"), S("ASI:One"), S("tool_calling")))
    metta.space().add_atom(E(S("capability"), S("ASI:One"), S("OpenAI_compatible_API")))
    metta.space().add_atom(E(S("capability"), S("Agentverse"), S("hosted_agents")))
    metta.space().add_atom(E(S("capability"), S("Agentverse"), S("agent_discovery")))
    metta.space().add_atom(E(S("capability"), S("Agentverse"), S("agent_deployment")))
    metta.space().add_atom(E(S("capability"), S("Agentverse"), S("mailbox")))
    metta.space().add_atom(E(S("capability"), S("Agentverse"), S("integrated_IDE")))
    metta.space().add_atom(E(S("capability"), S("Agentverse"), S("agent_search")))

    # Communication
    metta.space().add_atom(E(S("communication"), S("ctx.send"), S("async_delivery")))
    metta.space().add_atom(E(S("communication"), S("ctx.send"), S("with_acknowledgements")))
    metta.space().add_atom(E(S("communication"), S("ctx.send_and_receive"), S("request_response")))
    metta.space().add_atom(E(S("communication"), S("chat_protocol"), S("structured_messaging")))
    metta.space().add_atom(E(S("communication"), S("chat_protocol"), S("acknowledgements")))

    # Deployment (accurate uptime language — not "always running")
    metta.space().add_atom(E(S("deployment"), S("hosted_agents"), S("Agentverse_managed")))
    metta.space().add_atom(E(S("deployment"), S("hosted_agents"), S("active_while_started")))
    metta.space().add_atom(E(S("deployment"), S("local_agents"), S("self_hosted")))
    metta.space().add_atom(E(S("deployment"), S("local_agents"), S("full_library_access")))
    metta.space().add_atom(E(S("deployment"), S("mailbox_agents"), S("local_plus_Agentverse_inbox")))

    # Adapters (match current Adapters examples)
    metta.space().add_atom(E(S("adapter"), S("LangChain"), S("LangchainRegisterTool")))
    metta.space().add_atom(E(S("adapter"), S("LangGraph"), S("LangchainRegisterTool")))
    metta.space().add_atom(E(S("adapter"), S("CrewAI"), S("CrewaiRegisterTool")))
    metta.space().add_atom(E(S("adapter"), S("A2A_inbound"), S("external_A2A_clients")))
    metta.space().add_atom(E(S("adapter"), S("A2A_outbound"), S("A2A_as_uAgent")))

    # Blockchain: identity/registration vs hosted runtime
    metta.space().add_atom(E(S("blockchain"), S("Almanac"), S("agent_registry")))
    metta.space().add_atom(E(S("blockchain"), S("agent_identity"), S("on_chain_verifiable_address")))
    metta.space().add_atom(E(S("blockchain"), S("hosted_runtime"), S("managed_trusted_service")))

    # Current ASI:One models (https://docs.asi1.ai/documentation/models)
    metta.space().add_atom(E(S("specificInstance"), S("ASI:One"), S("asi1")))
    metta.space().add_atom(E(S("specificInstance"), S("ASI:One"), S("asi1-mini")))
    metta.space().add_atom(E(S("specificInstance"), S("ASI:One"), S("asi1-ultra")))
    metta.space().add_atom(E(S("capability"), S("asi1"), S("general_purpose_default")))
    metta.space().add_atom(E(S("capability"), S("asi1"), S("image_input")))
    metta.space().add_atom(E(S("capability"), S("asi1"), S("configurable_reasoning")))
    metta.space().add_atom(E(S("capability"), S("asi1-mini"), S("fastest_lightest")))
    metta.space().add_atom(E(S("capability"), S("asi1-mini"), S("high_volume_routing")))
    metta.space().add_atom(E(S("capability"), S("asi1-ultra"), S("most_capable")))
    metta.space().add_atom(E(S("capability"), S("asi1-ultra"), S("deep_research_and_review")))

    # Solutions (ValueAtom free text)
    metta.space().add_atom(
        E(
            S("solution"),
            S("create_uagent"),
            ValueAtom("pip install uagents; define Agent with name/seed/port; add handlers; run agent"),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("hosted_agent"),
            ValueAtom("use Agentverse IDE, write Python, click Start; agent is active while started"),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("local_agent"),
            ValueAtom("define port/endpoint, run locally, manage process uptime yourself"),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("mailbox_agent"),
            ValueAtom("set mailbox=True, run locally, Connect → Mailbox in Agentverse inspector"),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("agent_messaging"),
            ValueAtom("ctx.send for async (+ ChatAcknowledgement); ctx.send_and_receive for sync request/response"),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("chat_protocol"),
            ValueAtom("import ChatMessage/ChatAcknowledgement/TextContent; include chat_protocol_spec"),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("LangChain_integration"),
            ValueAtom("use LangchainRegisterTool from uagents_adapter; wrap agent function; see Adapters docs"),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("LangGraph_integration"),
            ValueAtom("wrap LangGraph executor; register with LangchainRegisterTool (same tool as LangChain examples)"),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("CrewAI_integration"),
            ValueAtom("use CrewaiRegisterTool from uagents_adapter; wrap crew handler"),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("search_agents"),
            ValueAtom(
                "POST https://agentverse.ai/v1/search/agents with auth + JSON filters; "
                "see Searching agents docs"
            ),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("mailbox_setup"),
            ValueAtom("enable Mailbox (older name: Mailroom) so unreachable agents queue messages until reconnect"),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("ASI:One_API_setup"),
            ValueAtom("create key at https://asi1.ai/dashboard/api-keys; use OpenAI-compatible base https://api.asi1.ai/v1"),
        )
    )
    metta.space().add_atom(
        E(
            S("solution"),
            S("model_selection"),
            ValueAtom("asi1 default; asi1-mini for speed/cost; asi1-ultra for hardest tasks — https://docs.asi1.ai/documentation/models"),
        )
    )

    # Considerations
    metta.space().add_atom(
        E(
            S("consideration"),
            S("hosted_agents"),
            ValueAtom("curated libraries only; active while started / inactive when stopped; plan limits apply; no public uptime SLA"),
        )
    )
    metta.space().add_atom(
        E(
            S("consideration"),
            S("local_agents"),
            ValueAtom("full library access; you manage process uptime and reachability"),
        )
    )
    metta.space().add_atom(
        E(
            S("consideration"),
            S("mailbox_agents"),
            ValueAtom("local process must stay running; Mailbox queues when briefly unreachable"),
        )
    )
    metta.space().add_atom(
        E(
            S("consideration"),
            S("ctx.send"),
            ValueAtom("async delivery; pair with ChatAcknowledgement in Chat Protocol; use send_and_receive when you need a reply"),
        )
    )
    metta.space().add_atom(
        E(
            S("consideration"),
            S("Agentverse"),
            ValueAtom("best-effort availability; hosted IDE is a trusted managed runtime, not trustless execution"),
        )
    )
    metta.space().add_atom(
        E(
            S("consideration"),
            S("ASI:One"),
            ValueAtom("API dependency and token costs; model choice trades latency, depth, and price"),
        )
    )
    metta.space().add_atom(
        E(
            S("consideration"),
            S("asi1-mini"),
            ValueAtom("fastest and lightest; trade depth and response length for speed"),
        )
    )
    metta.space().add_atom(
        E(
            S("consideration"),
            S("asi1-ultra"),
            ValueAtom("highest quality for hard tasks; expect higher latency and cost"),
        )
    )
    metta.space().add_atom(
        E(
            S("consideration"),
            S("Almanac"),
            ValueAtom("on-chain registry for discovery; registration has overhead"),
        )
    )

    # FAQ keys must match query_faq (not the raw user sentence) — one definition of ASI:One
    metta.space().add_atom(
        E(S("faq"), S("hi"), ValueAtom("Hello! I am a toy Fetch.ai/uAgents MeTTa assistant. How can I help?"))
    )
    metta.space().add_atom(
        E(
            S("faq"),
            S("what_is_fetchai"),
            ValueAtom(
                "Fetch.ai builds tools for autonomous agents: uAgents, Agentverse hosting/discovery, "
                "and ASI:One for LLM + agent routing."
            ),
        )
    )
    metta.space().add_atom(
        E(
            S("faq"),
            S("what_are_uagents"),
            ValueAtom(
                "uAgents are lightweight agent microservices for messaging, events, REST, and protocols "
                "in the Fetch.ai network."
            ),
        )
    )
    metta.space().add_atom(
        E(
            S("faq"),
            S("create_uagent"),
            ValueAtom(
                "Install uagents, define an Agent with a stable seed, add handlers, and run. "
                "See uAgent creation docs."
            ),
        )
    )
    metta.space().add_atom(
        E(
            S("faq"),
            S("hosted_vs_local"),
            ValueAtom(
                "Hosted agents run in the Agentverse IDE (curated packages, active while started). "
                "Local agents run on your machine with any packages; use Mailbox to stay reachable."
            ),
        )
    )
    metta.space().add_atom(
        E(
            S("faq"),
            S("mailbox_agent"),
            ValueAtom(
                "Set mailbox=True, run locally, open the inspector, Connect → Mailbox. "
                "Older docs may say Mailroom for the same inbox idea."
            ),
        )
    )
    metta.space().add_atom(
        E(
            S("faq"),
            S("what_is_asi_one"),
            ValueAtom(
                "ASI:One is an AI platform and LLM API for language, reasoning, coding, and finding "
                "Agentverse agents to help with tasks. Models: asi1, asi1-mini, asi1-ultra."
            ),
        )
    )
    metta.space().add_atom(
        E(
            S("faq"),
            S("asi_one_api_key"),
            ValueAtom("Create a key at https://asi1.ai/dashboard/api-keys and store it in .env as ASI_ONE_API_KEY."),
        )
    )
    metta.space().add_atom(
        E(
            S("faq"),
            S("which_asi_one_model"),
            ValueAtom(
                "Use asi1 by default; asi1-mini for fast/cheap high-volume work; asi1-ultra when quality "
                "beats latency. See https://docs.asi1.ai/documentation/models"
            ),
        )
    )
    metta.space().add_atom(
        E(
            S("faq"),
            S("what_is_agentverse"),
            ValueAtom(
                "Agentverse is a cloud platform to build, host, and discover agents. Hosted agents are "
                "active while started and inactive when stopped; availability is best-effort."
            ),
        )
    )
    metta.space().add_atom(
        E(
            S("faq"),
            S("search_agents"),
            ValueAtom(
                "POST https://agentverse.ai/v1/search/agents with Authorization and JSON filters. "
                "See Searching agents docs."
            ),
        )
    )
    metta.space().add_atom(
        E(
            S("faq"),
            S("langchain_uagents"),
            ValueAtom(
                "Yes — wrap your LangChain agent and register with LangchainRegisterTool from uagents_adapter."
            ),
        )
    )

Worked FAQ example: user asks How do I create a uAgent? → classifier keyword create_uagent → graph key create_uagent → seeded steps.

Limitation: the classifier extracts one keyword. Sample queries below use keys that match the seed graph.

Step 2: Implement MeTTa lookup​

Create metta/generalrag.py:

import re
from hyperon import MeTTa, E, S, ValueAtom

# Allow ASI:One model IDs (asi1-mini) and namespaced keys (ASI:One)
SYMBOL_PATTERN = re.compile(r"^[a-zA-Z0-9_:.-]+$")


def to_symbol(token: str):
    """Encode multi-word names; reject tokens that would break MeTTa."""
    if token is None:
        return None
    symbol = (
        str(token)
        .strip()
        .strip('"')
        .replace("'", "")
        .replace("\u2019", "")
        .replace(" ", "_")
    )
    if not SYMBOL_PATTERN.fullmatch(symbol):
        return None
    return symbol


def atom_to_str(atom) -> str:
    """Parse both Symbol and ValueAtom results."""
    try:
        obj = atom.get_object()
        if obj is not None and hasattr(obj, "value"):
            return str(obj.value)
    except Exception:
        pass
    return str(atom).strip('"')


class GeneralRAG:
    """MeTTa knowledge lookup (not embedding RAG)."""

    def __init__(self, metta_instance: MeTTa):
        self.metta = metta_instance

    def _run_match(self, relation: str, subject: str):
        symbol = to_symbol(subject)
        if not symbol:
            return []
        query_str = f"!(match &self ({relation} {symbol} $x) $x)"
        results = self.metta.run(query_str)
        if not results:
            return []
        values = []
        for row in results:
            if row and len(row) > 0:
                values.append(atom_to_str(row[0]))
        return list(dict.fromkeys(values))

    def query_capability(self, capability):
        return self._run_match("capability", capability)

    def get_solution(self, problem):
        return self._run_match("solution", problem)

    def get_consideration(self, topic):
        return self._run_match("consideration", topic)

    def query_faq(self, question_or_key):
        key = to_symbol(question_or_key)
        if not key:
            return None
        results = self._run_match("faq", key)
        return results[0] if results else None

    def get_specific_models(self, model: str):
        """Return specificInstance children (e.g. ASI:One → asi1, asi1-mini, asi1-ultra)."""
        return self._run_match("specificInstance", model)

    def query_all_specific_capabilities(self, model: str):
        """Return capabilities for each specificInstance under a parent model family."""
        parent = to_symbol(model)
        if not parent:
            return []
        query_str = (
            f"!(match &self "
            f"(, (specificInstance {parent} $specificInstance) "
            f"(capability $specificInstance $specificCapability)) "
            f"($specificInstance $specificCapability))"
        )
        results = self.metta.run(query_str)
        if not results:
            return []
        pairs = []
        for row in results:
            if row and len(row) >= 1:
                pairs.append(
                    atom_to_str(row[0])
                    if len(row) == 1
                    else " ".join(atom_to_str(a) for a in row)
                )
        return list(dict.fromkeys(pairs))

    def add_knowledge(self, relation_type, subject, object_value):
        rel = to_symbol(relation_type)
        subj = to_symbol(subject)
        if not rel or not subj or object_value is None:
            return "Skipped invalid knowledge"

        # Keep seed conventions: capability/adapter/deployment/communication/blockchain/specificInstance → S
        # solution/consideration/faq → ValueAtom
        symbol_relations = {
            "capability",
            "adapter",
            "deployment",
            "communication",
            "blockchain",
            "specificInstance",
        }
        if rel in symbol_relations:
            obj = to_symbol(object_value)
            if not obj:
                return "Skipped invalid object symbol"
            atom_obj = S(obj)
        else:
            atom_obj = ValueAtom(str(object_value))

        self.metta.space().add_atom(E(S(rel), S(subj), atom_obj))
        return f"Added {rel}: {subj} -> {object_value}"

Key Methods:

  • query_capability(): Features for a concept key
  • get_solution() / get_consideration(): Free-text ValueAtoms
  • query_faq(): FAQ by stable key (hi, create_uagent, what_is_asi_one), not the raw sentence
  • get_specific_models() / query_all_specific_capabilities(): Wired for ASI:One model questions
  • add_knowledge(): Same atom types as seed data (used only when LEARN=1)

Step 3: Query processing​

Create metta/utils.py. Fallback if not prompt: sits at function scope after all intent branches. Default path does not write LLM output into the graph.

import json
import os

from openai import OpenAI

from .generalrag import GeneralRAG, to_symbol

DISCLAIMER = (
    "Toy demo only. This MeTTa graph is a dated snapshot, not live Fetch.ai docs. "
    "Prefer Innovation Lab and ASI:One documentation for production guidance."
)
LEARN = os.getenv("LEARN") == "1"

# Map common classifier outputs onto seeded keys
KEYWORD_ALIASES = {
    "hosted": "hosted_agents",
    "local": "local_agents",
    "mailbox": "mailbox_agents",
    "uagent": "uAgent",
    "uagents": "uAgent",
    "asi1": "ASI:One",
    "asi:one": "ASI:One",
    "agentverse": "Agentverse",
    "langchain": "LangChain",
    "langgraph": "LangGraph",
    "crewai": "CrewAI",
    "create": "create_uagent",
    "fast": "asi1-mini",
    "model": "ASI:One",
}


class LLM:
    def __init__(self, api_key):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://api.asi1.ai/v1",
        )

    def create_completion(self, prompt, max_tokens=800):
        completion = self.client.chat.completions.create(
            messages=[{"role": "user", "content": prompt}],
            model="asi1",
            max_tokens=max_tokens,
        )
        return completion.choices[0].message.content


def normalize_keyword(keyword):
    if not keyword:
        return None
    raw = str(keyword).strip()
    aliased = KEYWORD_ALIASES.get(raw.lower().replace(" ", "_"), raw)
    return to_symbol(aliased)


def get_intent_and_keyword(query, llm):
    """Agent-side ASI:One call: classify intent and extract one keyword."""
    prompt = (
        f"Given the query: '{query}'\n"
        "Classify the intent as one of: 'capability', 'solution', 'consideration', 'faq', 'model', or 'unknown'.\n"
        "Extract the most relevant single keyword. Use snake_case for multi-word names "
        "(e.g. create_uagent, hosted_agents, model_selection).\n"
        "For greetings like Hi/Hello, intent=faq and keyword=hi.\n"
        "For 'how do I create a uAgent', intent=faq and keyword=create_uagent.\n"
        "For 'what is ASI:One', intent=faq and keyword=what_is_asi_one.\n"
        "For 'which ASI:One model for fast responses', intent=model and keyword=ASI:One.\n"
        "Return *only* JSON:\n"
        '{ "intent": "<classified_intent>", "keyword": "<extracted_keyword>" }'
    )
    response = llm.create_completion(prompt)
    try:
        cleaned = response.strip()
        if cleaned.startswith("```"):
            cleaned = "\n".join(cleaned.split("\n")[1:])
        if cleaned.endswith("```"):
            cleaned = "\n".join(cleaned.split("\n")[:-1])
        result = json.loads(cleaned.strip())
        return result["intent"], result.get("keyword")
    except (json.JSONDecodeError, KeyError):
        return "unknown", None


def generate_knowledge_response(query, intent, keyword, llm):
    """Optional LLM guess. Do not persist unless LEARN=1."""
    if intent == "capability":
        prompt = (
            f"Query: '{query}'\n"
            f"The concept '{keyword}' is not in the toy graph. Suggest one short capability phrase. "
            f"Return only that phrase. Remind it is unverified."
        )
    elif intent == "solution":
        prompt = (
            f"Query: '{query}'\n"
            f"No solution for '{keyword}' in the toy graph. Suggest concise steps. Return only that text."
        )
    elif intent == "consideration":
        prompt = (
            f"Query: '{query}'\n"
            f"No considerations for '{keyword}'. Suggest brief limitations. Return only that text."
        )
    elif intent in ("faq", "model"):
        prompt = (
            f"Query: '{query}'\n"
            "Provide a concise Fetch.ai/uAgents answer and say it is not from the verified toy graph. "
            "Return only the answer."
        )
    else:
        return None
    return llm.create_completion(prompt)


def _join(items, cap=8):
    items = [i for i in (items or []) if i]
    if not items:
        return "none in toy graph"
    return ", ".join(items[:cap])


def process_query(query, rag: GeneralRAG, llm: LLM):
    intent, keyword = get_intent_and_keyword(query, llm)
    keyword = normalize_keyword(keyword)
    prompt = ""

    if intent == "faq":
        faq_key = keyword or to_symbol(query)
        faq_answer = rag.query_faq(faq_key) if faq_key else None
        if faq_answer:
            prompt = (
                f"Query: '{query}'\n"
                f"FAQ Answer: '{faq_answer}'\n"
                "Humanize with a helpful developer tone. Keep the toy-demo meaning."
            )
        else:
            new_answer = generate_knowledge_response(query, intent, keyword, llm)
            if LEARN and faq_key and new_answer:
                rag.add_knowledge("faq", faq_key, new_answer)
            prompt = (
                f"Query: '{query}'\n"
                f"FAQ Answer (unverified LLM suggestion): '{new_answer}'\n"
                "Humanize helpfully. Do not present this as official Fetch.ai documentation."
            )
    elif intent == "model" and keyword:
        models = rag.get_specific_models(keyword)
        caps = rag.query_all_specific_capabilities(keyword)
        model_selection = rag.get_solution("model_selection")
        if models or caps or model_selection:
            prompt = (
                f"Query: '{query}'\n"
                f"Family: {keyword}\n"
                f"Models in toy graph: {_join(models)}\n"
                f"Per-model capabilities: {_join(caps, cap=12)}\n"
                f"Selection guidance: {_join(model_selection)}\n"
                "Recommend current IDs only (asi1 / asi1-mini / asi1-ultra). "
                "For fast responses, prefer asi1-mini."
            )
        else:
            suggestion = generate_knowledge_response(query, intent, keyword, llm)
            prompt = (
                f"Query: '{query}'\n"
                f"Model guidance (unverified LLM suggestion): {suggestion}\n"
                "Say this is not from the verified toy graph. Prefer docs.asi1.ai models page."
            )
    elif intent == "capability" and keyword:
        capabilities = rag.query_capability(keyword)
        if not capabilities:
            capability = generate_knowledge_response(query, intent, keyword, llm)
            if LEARN and capability:
                rag.add_knowledge("capability", keyword, capability)
            solutions = rag.get_solution(keyword) or []
            considerations = rag.get_consideration(keyword) or []
            prompt = (
                f"Query: '{query}'\n"
                f"Concept: {keyword}\n"
                f"Capabilities (unverified LLM suggestion): {capability}\n"
                f"Solutions in graph: {_join(solutions)}\n"
                f"Considerations: {_join(considerations)}\n"
                "Be explicit that suggestions outside the graph are unverified."
            )
        else:
            solutions = rag.get_solution(keyword) or []
            considerations = rag.get_consideration(keyword) or []
            prompt = (
                f"Query: '{query}'\n"
                f"Concept: {keyword}\n"
                f"Capabilities: {_join(capabilities)}\n"
                f"Solutions: {_join(solutions)}\n"
                f"Considerations: {_join(considerations)}\n"
                "Generate a concise, helpful response from the toy graph only."
            )
    elif intent == "solution" and keyword:
        solutions = rag.get_solution(keyword)
        if solutions:
            prompt = (
                f"Query: '{query}'\n"
                f"Problem: {keyword}\n"
                f"Solutions: {_join(solutions)}\n"
                "Provide a helpful suggestion from the toy graph only."
            )
        else:
            solution = generate_knowledge_response(query, intent, keyword, llm)
            if LEARN and solution:
                rag.add_knowledge("solution", keyword, solution)
            prompt = (
                f"Query: '{query}'\n"
                f"Problem: {keyword}\n"
                f"Solution (unverified LLM suggestion): {solution}\n"
                "Say this is not from the verified toy graph."
            )
    elif intent == "consideration" and keyword:
        considerations = rag.get_consideration(keyword) or []
        if considerations:
            prompt = (
                f"Query: '{query}'\n"
                f"Topic: {keyword}\n"
                f"Considerations: {_join(considerations)}\n"
                "Explain briefly from the toy graph."
            )
        else:
            consideration = generate_knowledge_response(query, intent, keyword, llm)
            if LEARN and consideration:
                rag.add_knowledge("consideration", keyword, consideration)
            prompt = (
                f"Query: '{query}'\n"
                f"Topic: {keyword}\n"
                f"Considerations (unverified LLM suggestion): {consideration}\n"
                "Do not present this as official product policy."
            )

    if not prompt:
        prompt = (
            f"Query: '{query}'\n"
            "No specific info found in the toy graph. Offer general Fetch.ai/uAgents assistance "
            "and point to Innovation Lab docs."
        )

    prompt += (
        f"\nAlways start the answer with: {DISCLAIMER}\n"
        "Then give the helpful content. Do not invent obsolete ASI:One model IDs "
        "(no asi1-fast / asi1-extended / asi1-agentic / asi1-graph)."
    )
    response = llm.create_completion(prompt, max_tokens=800)
    text = (response or "").strip()
    if DISCLAIMER.lower() not in text.lower():
        text = f"{DISCLAIMER}\n\n{text}"
    return {"selected_question": query, "humanized_answer": text}

Intent Classification (runs in the agent, after Chat Protocol receives text):

  • capability: features for a concept (uAgent, Agentverse, ASI:One)
  • solution: how-to keys (create_uagent, search_agents, model_selection)
  • consideration: limits for a topic (hosted_agents, asi1-mini)
  • faq: keyed FAQs (hi, what_is_asi_one, create_uagent)
  • model: lists specificInstance children and per-model capabilities

Step 4: Configure Agent​

Create agent.py at the project root (not inside metta/):

from datetime import datetime, timezone
from uuid import uuid4
import os
import sys

from dotenv import load_dotenv
from uagents import Context, Protocol, Agent
from hyperon import MeTTa

from uagents_core.contrib.protocols.chat import (
    ChatAcknowledgement,
    ChatMessage,
    EndSessionContent,
    StartSessionContent,
    TextContent,
    chat_protocol_spec,
)

from metta.generalrag import GeneralRAG
from metta.knowledge import initialize_knowledge_graph
from metta.utils import LLM, process_query, DISCLAIMER

load_dotenv()

api_key = os.getenv("ASI_ONE_API_KEY")
agent_seed = os.getenv("AGENT_SEED")

if not api_key:
    print("Missing ASI_ONE_API_KEY. Create a key at https://asi1.ai/dashboard/api-keys and put it in .env")
    sys.exit(1)

if not agent_seed:
    print("Missing AGENT_SEED. Set a unique local seed in .env (do not commit secrets).")
    sys.exit(1)

agent = Agent(
    name="Fetch.ai Knowledge MeTTa Assistant",
    seed=agent_seed,
    port=8005,
    mailbox=True,
    publish_agent_details=True,
)


def create_text_chat(text: str, end_session: bool = False) -> ChatMessage:
    content = [TextContent(type="text", text=text)]
    if end_session:
        content.append(EndSessionContent(type="end-session"))
    return ChatMessage(
        timestamp=datetime.now(timezone.utc),
        msg_id=uuid4(),
        content=content,
    )


metta = MeTTa()
initialize_knowledge_graph(metta)
rag = GeneralRAG(metta)
llm = LLM(api_key=api_key)

chat_proto = Protocol(spec=chat_protocol_spec)


@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
    ctx.storage.set(str(ctx.session), sender)
    await ctx.send(
        sender,
        ChatAcknowledgement(
            timestamp=datetime.now(timezone.utc),
            acknowledged_msg_id=msg.msg_id,
        ),
    )

    for item in msg.content:
        if isinstance(item, StartSessionContent):
            ctx.logger.info(f"Got a start session message from {sender}")
            continue
        elif isinstance(item, TextContent):
            user_query = item.text.strip()
            ctx.logger.info(f"Got a Fetch.ai/uAgents query from {sender}: {user_query}")
            try:
                response = process_query(user_query, rag, llm)
                answer_text = response.get(
                    "humanized_answer",
                    f"{DISCLAIMER}\n\nI could not process your query.",
                )
                await ctx.send(sender, create_text_chat(answer_text))
            except Exception as e:
                ctx.logger.error(f"Error processing Fetch.ai/uAgents query: {e}")
                await ctx.send(
                    sender,
                    create_text_chat(
                        f"{DISCLAIMER}\n\nI hit an error processing that query. Please try again."
                    ),
                )
        else:
            ctx.logger.info(f"Got unexpected content from {sender}")


@chat_proto.on_message(ChatAcknowledgement)
async def handle_ack(ctx: Context, sender: str, msg: ChatAcknowledgement):
    ctx.logger.info(
        f"Got an acknowledgement from {sender} for {msg.acknowledged_msg_id}"
    )


agent.include(chat_proto, publish_manifest=True)

if __name__ == "__main__":
    agent.run()

Agent Features:

  • Toy MeTTa lookup only — not live product documentation
  • Every reply includes a snapshot / prefer official docs disclaimer
  • Does not persist unverified LLM output into the graph unless LEARN=1
  • Agent LLM (ASI:One) classifies intent after Chat Protocol receives text
  • Compatible with ASI:One via Chat Protocol and Agentverse mailbox (mailbox=True)

Detailed Working (Step-by-Step)​

  1. User sends a query through ASI:One chat (or Inspector chat).
  2. Chat Protocol handler receives TextContent.
  3. The agent calls ASI:One (get_intent_and_keyword) to classify intent and one keyword.
  4. GeneralRAG runs MeTTa match queries on the toy graph.
  5. Reply is humanized, disclaimer is prepended, and Chat Protocol sends a string (not a dict).

Testing and Deployment​

Local Testing (mailbox)​

Numbered steps matching current uAgents + Agentverse. See also Mailbox agents and uAgent creation.

  1. Log in to Agentverse.

  2. From project/, with venv active and .env set:

    python3 agent.py

    Windows: py -3.10 agent.py.

  3. In the console, copy the inspector URL (it includes your agent address). Expected lines look like:

    INFO:     [Fetch.ai Knowledge MeTTa Assistant]: Starting agent with address: agent1q...
    INFO:     [Fetch.ai Knowledge MeTTa Assistant]: Agent inspector available at https://agentverse.ai/inspect/?uri=http%3A//127.0.0.1%3A8005&address=agent1q...
    INFO:     [Fetch.ai Knowledge MeTTa Assistant]: Starting mailbox client for https://agentverse.ai
    INFO:     [mailbox]: Successfully registered as mailbox agent in Agentverse

    If you see Missing ASI_ONE_API_KEY, stop and fix .env — the agent exits before agent.run().

  4. Open the inspector URL while logged in. Choose Connect → Mailbox (Agentverse issues the mailbox token; you do not paste Python import mailbox).

  5. Use Chat with Agent on the Inspector/profile, or continue to ASI:One below. Keep agent.py running.

Sample queries (aligned with one-keyword lookup)​

  • Hi → FAQ key hi (greeting)
  • How do I create a uAgent? → FAQ key create_uagent (seeded install steps)
  • What is the difference between hosted and local agents? → FAQ key hosted_vs_local
  • How do agents communicate with each other? → solution key agent_messaging / capability around chat_protocol
  • How can I integrate LangChain with uAgents? → FAQ key langchain_uagents
  • Which ASI:One model is best for fast responses? → intent model → asi1-mini among asi1 / asi1-mini / asi1-ultra
  • What is ASI:One? → single FAQ key what_is_asi_one

Query your agent from ASI:One​

ASI:One discovers mailbox agents that are running, registered, and using Chat Protocol. README/handle tips: Searching agents. Chat UI: ASI:One Chat.

  1. Copy the agent address from the console (agent1q...). Optionally set a handle on the Agentverse profile.
  2. Open ASI:One, sign in with Google or the ASI:One wallet, and start a new chat.
  3. Toggle Agents so ASI:One can call Agentverse agents.
  4. Paste the address or @handle and send a sample query such as How do I create a uAgent?
  5. Expect a reply that starts with the toy-demo disclaimer and mentions pip install / Agent / handlers from the seeded FAQ. The local console should log the incoming chat message.

Expected output​

Startup (shape of logs; address is unique to your AGENT_SEED):

INFO:     [Fetch.ai Knowledge MeTTa Assistant]: Starting agent with address: agent1q...
INFO:     [Fetch.ai Knowledge MeTTa Assistant]: Agent inspector available at https://agentverse.ai/inspect/?uri=...
INFO:     [mailbox]: Successfully registered as mailbox agent in Agentverse

Example chat:

  • You: How do I create a uAgent?
  • Agent: Starts with Toy demo only... then, from the toy graph, summarizes install → define Agent with seed → handlers → run.
  • You: Which ASI:One model is best for fast responses?
  • Agent: Recommends asi1-mini among current models asi1 / asi1-mini / asi1-ultra (does not mention obsolete model IDs).

Notes​

  • Shared MeTTa + uAgent skeleton with Medical and Financial pages; swap the domain graph, keep the package layout and LEARN gate.
  • For Agentverse uptime, Mailbox naming, Search API, and adapters, trust Agentverse Overview, Searching agents, and Adapters — not this toy graph.
  • Do not set LEARN=1 if you care about inventing Fetch.ai “facts” into the graph.