Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ Agent executes tools → LLM inference → Post-process response → Return Agen
- **agent/**: Agent orchestration - `agent_executors.py` creates LangGraph ReAct agents, `tools.py` defines agent tools, `prompts.py` loads Jinja2 templates
- **server/**: FastAPI app in `fastapi_server.py`, auth, validation, activity tracking
- **onchain/**: Blockchain data layer - `pools/` for DeFi protocol abstraction, `tokens/` for metadata, `portfolio/` for wallet analysis, `analytics/` for metrics
- **templates/**: Jinja2 prompt templates for agents (`analyst_agent.jinja2`, `investor_agent.jinja2`)
- **templates/**: Jinja2 prompt templates for agents (`analytics_agent.jinja2`, `investor_agent.jinja2`)
- **api/api_types.py**: All Pydantic models (Token, Pool, Portfolio, Message types)

### Protocol System
Expand Down
1 change: 0 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,6 @@ server/ # Flask server exposing the API
static/ # Static assets for web interface
templates/ # LLM prompt templates for agent
testclient/ # Client for testing the API
testutils/ # Utility functions for testing
```

### Agents
Expand Down
16 changes: 8 additions & 8 deletions agent/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -71,19 +71,19 @@ async def search_token(
token: str, chain: Optional[str] = None
) -> Optional[TokenMetadata]:
"""Search for a token by name or symbol. Returns metadata for the first token found."""
token: Optional[TokenMetadata] = await token_metadata_repo.search_token(
result: Optional[TokenMetadata] = await token_metadata_repo.search_token(
token, chain
)
if not token:
if not result:
return "No token found."

return {
"id": f"{token.chain}:{token.address}",
"address": token.address,
"name": token.name,
"symbol": token.symbol,
"price_usd": token.price,
"chain": token.chain,
"id": f"{result.chain}:{result.address}",
"address": result.address,
"name": result.name,
"symbol": result.symbol,
"price_usd": result.price,
"chain": result.chain,
}

return [
Expand Down
56 changes: 56 additions & 0 deletions test_search_token.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
"""
Test to verify search_token function does not shadow the input parameter.
"""
import unittest
from unittest.mock import AsyncMock, MagicMock


class TestSearchToken(unittest.IsolatedAsyncioTestCase):
async def test_search_token_no_variable_shadowing(self):
"""
Ensure search_token uses 'result' variable instead of shadowing
the 'token' string parameter with a TokenMetadata object.
"""
# Mock the token metadata repo
mock_repo = MagicMock()
mock_token = MagicMock()
mock_token.chain = "solana"
mock_token.address = "So111111111"
mock_token.name = "Wrapped SOL"
mock_token.symbol = "SOL"
mock_token.price = 150.0
mock_repo.search_token = AsyncMock(return_value=mock_token)

# Import and create the toolkit
from agent.tools import create_analytics_agent_toolkit
toolkit = create_analytics_agent_toolkit(mock_repo)

# Find the search_token tool
search_tool = next(t for t in toolkit if t.name == "search_token")

# Call with a string query
result = await search_tool.ainvoke({"token": "SOL", "chain": "solana"})

# Verify result is correct
self.assertEqual(result["symbol"], "SOL")
self.assertEqual(result["name"], "Wrapped SOL")
self.assertEqual(result["chain"], "solana")

async def test_search_token_not_found(self):
"""
Ensure search_token returns correct message when token not found.
"""
mock_repo = MagicMock()
mock_repo.search_token = AsyncMock(return_value=None)

from agent.tools import create_analytics_agent_toolkit
toolkit = create_analytics_agent_toolkit(mock_repo)

search_tool = next(t for t in toolkit if t.name == "search_token")
result = await search_tool.ainvoke({"token": "UNKNOWN"})

self.assertEqual(result, "No token found.")


if __name__ == "__main__":
unittest.main()