Skip to content
Closed
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
66 changes: 66 additions & 0 deletions skills/deckofcards/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
"""Deck of Cards skills."""

from typing import TypedDict

from abstracts.skill import SkillStoreABC
from skills.base import SkillConfig, SkillState
from skills.deckofcards.add_to_pile import AddToPile
from skills.deckofcards.base import DeckOfCardsBaseSkill
from skills.deckofcards.draw_cards import DrawCards
from skills.deckofcards.draw_from_pile import DrawFromPile
from skills.deckofcards.list_piles import ListPiles
from skills.deckofcards.shuffle_deck import ShuffleDeck

# Cache skills at the system level, because they are stateless
_cache: dict[str, DeckOfCardsBaseSkill] = {}


class SkillStates(TypedDict):
shuffle_deck: SkillState
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Or maybe you just missing others here?



class Config(SkillConfig):
"""Configuration for Deck of Cards skills."""

states: SkillStates


async def get_skills(
config: "Config",
is_private: bool,
store: SkillStoreABC,
**_,
) -> list[DeckOfCardsBaseSkill]:
"""Get all Deck of Cards skills."""
available_skills = []

# Include skills based on their state
for skill_name, state in config["states"].items():
if state == "disabled":
continue
elif state == "public" or (state == "private" and is_private):
available_skills.append(skill_name)

# Get each skill using the cached getter
return [get_deckofcards_skill(name, store) for name in available_skills]


def get_deckofcards_skill(
name: str,
store: SkillStoreABC,
) -> DeckOfCardsBaseSkill:
"""Get a Deck of Cards skill by name."""
if name not in _cache:
if name == "shuffle_deck":
_cache[name] = ShuffleDeck(skill_store=store)
elif name == "draw_cards":
_cache[name] = DrawCards(skill_store=store)
elif name == "add_to_pile":
_cache[name] = AddToPile(skill_store=store)
elif name == "draw_from_pile":
_cache[name] = DrawFromPile(skill_store=store)
elif name == "list_piles":
_cache[name] = ListPiles(skill_store=store)
else:
raise ValueError(f"Unknown Deck of Cards skill: {name}")
return _cache[name]
58 changes: 58 additions & 0 deletions skills/deckofcards/add_to_pile.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
from typing import List

import httpx

from .base import DeckOfCardsBaseSkill


class AddToPile(DeckOfCardsBaseSkill):
"""Skill to add cards to a named pile."""

name = "add_to_pile"
description = "Add specified cards to a named pile in the deck"

async def _arun(self, deck_id: str, pile_name: str, cards: List[str]) -> str:
"""
Add cards to a named pile.

Args:
deck_id: ID of the deck
pile_name: Name of the pile to add cards to
cards: List of card codes to add (e.g., ["AS", "2S"])

Returns:
str: Formatted response containing pile information
"""
try:
# Convert list of cards to comma-separated string
cards_str = ",".join(cards)

async with httpx.AsyncClient() as client:
response = await client.get(
f"https://deckofcardsapi.com/api/deck/{deck_id}/pile/{pile_name}/add/",
params={"cards": cards_str},
)
response.raise_for_status()
data = response.json()

if not data["success"]:
return "Error: Failed to add cards to pile"

result = [
"Cards Added to Pile Successfully",
f"Deck ID: {data['deck_id']}",
f"Cards Remaining in Deck: {data['remaining']}",
"",
"Pile Status:",
]

# Add information about each pile
for pile_name, pile_info in data["piles"].items():
result.append(f"- {pile_name}: {pile_info['remaining']} cards")

return "\n".join(result)

except httpx.HTTPError as e:
return f"Error adding cards to pile: {str(e)}"
except Exception as e:
return f"Unexpected error: {str(e)}"
21 changes: 21 additions & 0 deletions skills/deckofcards/base.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
from typing import Type

from pydantic import BaseModel, Field

from abstracts.skill import SkillStoreABC
from skills.base import IntentKitSkill


class DeckOfCardsBaseSkill(IntentKitSkill):
"""Base class for Deck of Cards API tools."""

name: str = Field(description="The name of the tool")
description: str = Field(description="A description of what the tool does")
args_schema: Type[BaseModel]
skill_store: SkillStoreABC = Field(
description="The skill store for persisting data"
)

@property
def category(self) -> str:
return "deckofcards"
Binary file added skills/deckofcards/deckofcards.png
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
57 changes: 57 additions & 0 deletions skills/deckofcards/draw_cards.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
import httpx

from .base import DeckOfCardsBaseSkill


class DrawCards(DeckOfCardsBaseSkill):
"""Skill to draw cards from a deck."""

name = "draw_cards"
description = "Draw a specified number of cards from a deck"

async def _arun(self, deck_id: str, count: int = 1) -> str:
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If there are parameters, you need to declare them in args_schema; otherwise, the AI will not call it. It seems there are still many problems; please complete it and pass the local test.

"""
Draw cards from a deck.

Args:
deck_id: ID of the deck to draw from
count: Number of cards to draw (default: 1)
params: Additional parameters (unused)

Returns:
str: Formatted response containing drawn cards information
"""
try:
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://deckofcardsapi.com/api/deck/{deck_id}/draw/",
params={"count": count},
)
response.raise_for_status()
data = response.json()

if not data["success"]:
return "Error: Failed to draw cards"

result = [
"Cards Drawn Successfully",
f"Deck ID: {data['deck_id']}",
f"Cards Remaining: {data['remaining']}",
"",
"Drawn Cards:",
]

for card in data["cards"]:
result.extend(
[
f"- {card['value']} of {card['suit']}",
f" Image: {card['image']}",
]
)

return "\n".join(result)

except httpx.HTTPError as e:
return f"Error drawing cards: {str(e)}"
except Exception as e:
return f"Unexpected error: {str(e)}"
69 changes: 69 additions & 0 deletions skills/deckofcards/draw_from_pile.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
from typing import List

import httpx

from .base import DeckOfCardsBaseSkill


class DrawFromPile(DeckOfCardsBaseSkill):
"""Skill to draw cards from a named pile."""

name = "draw_from_pile"
description = "Draw specified cards from a named pile in the deck"

async def _arun(self, deck_id: str, pile_name: str, cards: List[str]) -> str:
"""
Draw cards from a named pile.

Args:
deck_id: ID of the deck
pile_name: Name of the pile to draw from
cards: List of card codes to draw (e.g., ["AS", "2S"])

Returns:
str: Formatted response containing drawn cards and pile information
"""
try:
# Convert list of cards to comma-separated string
cards_str = ",".join(cards)

async with httpx.AsyncClient() as client:
response = await client.get(
f"https://deckofcardsapi.com/api/deck/{deck_id}/pile/{pile_name}/draw/",
params={"cards": cards_str},
)
response.raise_for_status()
data = response.json()

if not data["success"]:
return "Error: Failed to draw cards from pile"

result = [
"Cards Drawn from Pile Successfully",
f"Deck ID: {data['deck_id']}",
f"Cards Remaining in Deck: {data['remaining']}",
"",
"Pile Status:",
]

# Add information about each pile
for pile_name, pile_info in data["piles"].items():
result.append(f"- {pile_name}: {pile_info['remaining']} cards")

# Add information about drawn cards
if "cards" in data:
result.extend(["", "Drawn Cards:"])
for card in data["cards"]:
result.extend(
[
f"- {card['value']} of {card['suit']}",
f" Image: {card['image']}",
]
)

return "\n".join(result)

except httpx.HTTPError as e:
return f"Error drawing cards from pile: {str(e)}"
except Exception as e:
return f"Unexpected error: {str(e)}"
62 changes: 62 additions & 0 deletions skills/deckofcards/list_piles.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
import httpx

from .base import DeckOfCardsBaseSkill


class ListPiles(DeckOfCardsBaseSkill):
"""Skill to list all piles and their contents in a deck."""

name = "list_piles"
description = "View existing piles and their contents in the deck"

async def _arun(self, deck_id: str) -> str:
"""
List all piles and their contents.

Args:
deck_id: ID of the deck

Returns:
str: Formatted response containing pile information and contents
"""
try:
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://deckofcardsapi.com/api/deck/{deck_id}/pile/list/"
)
response.raise_for_status()
data = response.json()

if not data["success"]:
return "Error: Failed to list piles"

result = [
"Piles Listed Successfully",
f"Deck ID: {data['deck_id']}",
f"Cards Remaining in Deck: {data['remaining']}",
"",
"Pile Contents:",
]

# Add information about each pile
for pile_name, pile_info in data["piles"].items():
result.append(f"\n{pile_name}:")
result.append(f"- Cards Remaining: {pile_info['remaining']}")

# Add card details if available
if "cards" in pile_info:
result.append("- Cards:")
for card in pile_info["cards"]:
result.extend(
[
f" * {card['value']} of {card['suit']}",
f" Image: {card['image']}",
]
)

return "\n".join(result)

except httpx.HTTPError as e:
return f"Error listing piles: {str(e)}"
except Exception as e:
return f"Unexpected error: {str(e)}"
Loading