Skip to content
Merged
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
33 changes: 21 additions & 12 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
[![PyPI version](https://img.shields.io/pypi/v/agentscore-py.svg)](https://pypi.org/project/agentscore-py/)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)

Python client for the [AgentScore](https://agentscore.sh) trust and reputation API. Score, verify, and assess AI agent wallets in the [x402](https://github.com/coinbase/x402) payment ecosystem and [ERC-8004](https://eips.ethereum.org/EIPS/eip-8004) agent registry.
Python client for the [AgentScore](https://agentscore.sh) trust and reputation API.

## Install

Expand All @@ -16,29 +16,40 @@ pip install agentscore-py
```python
from agentscore import AgentScore

client = AgentScore(api_key="ask_...")
client = AgentScore(api_key="as_live_...")

# Free reputation lookup
# Look up cached reputation (free)
rep = client.get_reputation("0x1234...")
print(rep["grade"], rep["score"])
print(rep["score"]["value"], rep["score"]["grade"])

# Trust decision with policy
decision = client.get_decision("0x1234...", min_grade="C", min_transactions=5)
print(decision["decision"]["allow"])
# Filter to a specific chain
base_rep = client.get_reputation("0x1234...", chain="base")

# On-the-fly assessment with policy (paid)
result = client.assess("0x1234...", policy={"min_grade": "B", "min_score": 35})
print(result["decision"], result["decision_reasons"])

# Browse agents
agents = client.get_agents(chain="base", limit=10)
print(len(agents["items"]), agents["count"])

# Ecosystem stats
stats = client.get_stats()
print(stats["erc8004"]["known_agents"])
```

### Async

```python
async with AgentScore(api_key="ask_...") as client:
async with AgentScore(api_key="as_live_...") as client:
rep = await client.aget_reputation("0x1234...")
print(rep["grade"])
result = await client.aassess("0x1234...", policy={"min_grade": "B"})
```

### Context Manager

```python
with AgentScore(api_key="ask_...") as client:
with AgentScore(api_key="as_live_...") as client:
rep = client.get_reputation("0x1234...")
```

Expand All @@ -64,8 +75,6 @@ except AgentScoreError as e:
## Documentation

- [API Reference](https://docs.agentscore.sh)
- [ERC-8004 Standard](https://eips.ethereum.org/EIPS/eip-8004)
- [x402 Protocol](https://github.com/coinbase/x402)

## License

Expand Down
2 changes: 2 additions & 0 deletions agentscore/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
EntityType,
Grade,
ReputationResponse,
ReputationResponseFull,
ReputationStatus,
StatsERC8004,
StatsPayments,
Expand All @@ -29,6 +30,7 @@
"EntityType",
"Grade",
"ReputationResponse",
"ReputationResponseFull",
"ReputationStatus",
"StatsERC8004",
"StatsPayments",
Expand Down
26 changes: 13 additions & 13 deletions agentscore/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
AgentsListResponse,
AssessResponse,
DecisionPolicy,
ReputationResponse,
ReputationResponseFull,
StatsResponse,
)

Expand Down Expand Up @@ -86,10 +86,10 @@ def _handle_response(self, response: httpx.Response) -> dict:

# --- Sync methods ---

def get_reputation(self, address: str, chain: str = "base") -> ReputationResponse:
"""Get cached reputation for an address (free, read-only)."""
def get_reputation(self, address: str, chain: str | None = None) -> ReputationResponseFull:
"""Get cached reputation for an address (free, read-only). Optionally filter by chain."""
params: dict[str, str] = {}
if chain != "base":
if chain:
params["chain"] = chain
client = self._get_sync_client()
response = client.get(f"/v1/reputation/{address}", params=params)
Expand All @@ -98,13 +98,13 @@ def get_reputation(self, address: str, chain: str = "base") -> ReputationRespons
def assess(
self,
address: str,
chain: str = "base",
chain: str | None = None,
refresh: bool = False,
policy: DecisionPolicy | None = None,
) -> AssessResponse:
"""Assess a wallet (paid, writes score on-the-fly)."""
body: dict[str, Any] = {"address": address}
if chain != "base":
if chain:
body["chain"] = chain
if refresh:
body["refresh"] = True
Expand All @@ -115,7 +115,7 @@ def assess(
return self._handle_response(response)

def get_agents(self, **filters: Any) -> AgentsListResponse:
"""Browse ERC-8004 agents (free)."""
"""Browse agents (free)."""
params = {k: str(v).lower() if isinstance(v, bool) else str(v) for k, v in filters.items() if v is not None}
client = self._get_sync_client()
response = client.get("/v1/agents", params=params)
Expand All @@ -129,10 +129,10 @@ def get_stats(self) -> StatsResponse:

# --- Async methods ---

async def aget_reputation(self, address: str, chain: str = "base") -> ReputationResponse:
"""Get cached reputation for an address (free, read-only)."""
async def aget_reputation(self, address: str, chain: str | None = None) -> ReputationResponseFull:
"""Get cached reputation for an address (free, read-only). Optionally filter by chain."""
params: dict[str, str] = {}
if chain != "base":
if chain:
params["chain"] = chain
client = self._get_async_client()
response = await client.get(f"/v1/reputation/{address}", params=params)
Expand All @@ -141,13 +141,13 @@ async def aget_reputation(self, address: str, chain: str = "base") -> Reputation
async def aassess(
self,
address: str,
chain: str = "base",
chain: str | None = None,
refresh: bool = False,
policy: DecisionPolicy | None = None,
) -> AssessResponse:
"""Assess a wallet (paid, writes score on-the-fly)."""
body: dict[str, Any] = {"address": address}
if chain != "base":
if chain:
body["chain"] = chain
if refresh:
body["refresh"] = True
Expand All @@ -158,7 +158,7 @@ async def aassess(
return self._handle_response(response)

async def aget_agents(self, **filters: Any) -> AgentsListResponse:
"""Browse ERC-8004 agents (free)."""
"""Browse agents (free)."""
params = {k: str(v).lower() if isinstance(v, bool) else str(v) for k, v in filters.items() if v is not None}
client = self._get_async_client()
response = await client.get("/v1/agents", params=params)
Expand Down
31 changes: 20 additions & 11 deletions agentscore/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,21 +32,10 @@ class Score(TypedDict):
version: str


class ERC8004Identity(TypedDict, total=False):
chain: str
token_id: int
registry_contract: str | None
name: str | None
description: str | None
metadata_quality: str | None
endpoint_count: int


class Identity(TypedDict):
ens_name: str | None
website_url: str | None
github_url: str | None
erc8004: ERC8004Identity | None


class Activity(TypedDict):
Expand All @@ -65,6 +54,21 @@ class Activity(TypedDict):
last_verified_tx_at: str | None


class OperatorScore(TypedDict):
score: int
grade: Grade
agent_count: int
chains_active: list[str]


class AgentSummary(TypedDict):
token_id: int
chain: str
name: str | None
score: int
grade: Grade


class ReputationResponse(TypedDict):
subject: Subject
classification: Classification
Expand All @@ -77,6 +81,11 @@ class ReputationResponse(TypedDict):
updated_at: str | None


class ReputationResponseFull(ReputationResponse, total=False):
operator_score: OperatorScore
agents: list[AgentSummary]


class DecisionPolicy(TypedDict, total=False):
min_grade: Grade
min_score: int
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "hatchling.build"

[project]
name = "agentscore-py"
version = "1.0.4"
version = "1.1.0"
description = "Python client for the AgentScore trust and reputation API"
readme = "README.md"
license = "MIT"
Expand Down
15 changes: 7 additions & 8 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,24 +96,23 @@ def test_get_reputation_success():


@respx.mock
def test_get_reputation_with_non_default_chain():
def test_get_reputation_no_chain_param():
route = respx.get(f"{BASE_URL}/v1/reputation/{ADDRESS}").mock(
return_value=httpx.Response(200, json=REPUTATION_PAYLOAD)
)
client = AgentScore(api_key=API_KEY)
client.get_reputation(ADDRESS, chain="ethereum")
assert "chain=ethereum" in str(route.calls.last.request.url)
client.get_reputation(ADDRESS)
assert "chain" not in str(route.calls.last.request.url)


@respx.mock
def test_get_reputation_default_chain_omitted_from_params():
def test_get_reputation_with_chain():
route = respx.get(f"{BASE_URL}/v1/reputation/{ADDRESS}").mock(
return_value=httpx.Response(200, json=REPUTATION_PAYLOAD)
)
client = AgentScore(api_key=API_KEY)
client.get_reputation(ADDRESS, chain="base")
# chain=base should NOT be included in the query string
assert "chain" not in str(route.calls.last.request.url)
assert "chain=base" in str(route.calls.last.request.url)


@respx.mock
Expand Down Expand Up @@ -201,10 +200,10 @@ def test_assess_with_non_default_chain():


@respx.mock
def test_assess_default_chain_omitted_from_body():
def test_assess_no_chain_omits_from_body():
route = respx.post(f"{BASE_URL}/v1/assess").mock(return_value=httpx.Response(200, json=ASSESS_PAYLOAD))
client = AgentScore(api_key=API_KEY)
client.assess(ADDRESS, chain="base")
client.assess(ADDRESS)
body = json.loads(route.calls.last.request.content)
assert "chain" not in body

Expand Down
Loading
Loading