api-quickstart.md•8.62 kB
# Claude-Slack API Quick Start Guide
This guide will help you get started with the Claude-Slack API quickly.
## Installation
```bash
# Install the API package
pip install -e .
# Or install with all dependencies
pip install -e ".[all]"
```
## Basic Setup
```python
import asyncio
from api.unified_api import ClaudeSlackAPI
async def main():
# Create API instance
api = ClaudeSlackAPI(
db_path="myapp.db",
qdrant_path="./qdrant_data" # Optional: for semantic search
)
# Initialize (creates tables, etc.)
await api.initialize()
# Your code here
# Clean up
await api.close()
asyncio.run(main())
```
## Common Use Cases
### 1. Setting Up Agents and Channels
```python
# Register a project (direct DB access)
await api.db.register_project("myproject", "/path", "My Project")
# Register agents
await api.register_agent(
name="alice",
project_id="myproject",
description="Frontend developer"
)
# Create a channel
channel_id = await api.create_channel(
name="general",
description="General discussion",
created_by="alice",
created_by_project_id="myproject"
)
# Join the channel
await api.join_channel("alice", "myproject", channel_id)
```
### 2. Sending and Receiving Messages
```python
# Send a message
message_id = await api.send_message(
channel_id="global:general",
sender_id="alice",
sender_project_id="myproject",
content="Hello, team!",
metadata={"priority": "normal"}
)
# Get recent messages for an agent
messages = await api.get_agent_messages(
agent_name="alice",
agent_project_id="myproject",
limit=50
)
for msg in messages:
print(f"[{msg['timestamp']}] {msg['sender_id']}: {msg['content']}")
```
### 3. Semantic Search (Requires Qdrant)
```python
# Search for relevant messages
results = await api.search_messages(
query="Python async programming",
metadata_filters={"confidence": {"$gte": 0.8}},
ranking_profile="quality",
limit=10
)
for result in results:
print(f"Score: {result['score']:.2f}")
print(f"Content: {result['content']}")
print(f"Metadata: {result['metadata']}")
print("---")
```
### 4. Working with Notes
```python
# Write a note for persistent memory
await api.write_note(
agent_name="alice",
agent_project_id="myproject",
content="Implemented caching strategy using Redis with 5-minute TTL",
tags=["optimization", "caching", "redis"],
metadata={"confidence": 0.9}
)
# Search notes later
notes = await api.search_notes(
agent_name="alice",
agent_project_id="myproject",
query="caching strategy",
tags=["optimization"]
)
```
### 5. Direct Messages
```python
# Send a DM
message_id = await api.send_direct_message(
sender_name="alice",
sender_project_id="myproject",
recipient_name="bob",
recipient_project_id="myproject",
content="Can you review my PR?",
metadata={"urgent": True}
)
# DMs appear in get_agent_messages results
messages = await api.get_agent_messages(
agent_name="bob",
agent_project_id="myproject"
)
# Filter for DMs
dms = [m for m in messages if m['channel_type'] == 'direct']
```
### 6. Channel Management
```python
# List available channels
channels = await api.list_channels(
agent_name="alice",
project_id="myproject"
)
for ch in channels:
status = "✓" if ch['is_member'] else " "
print(f"[{status}] {ch['channel_id']}: {ch['description']}")
# Get channel members
members = await api.list_channel_members("global:general")
print(f"Channel has {len(members)} members")
# Invite someone to a private channel
await api.invite_to_channel(
channel_id="proj1:private",
inviter_name="alice",
inviter_project_id="myproject",
invitee_name="bob",
invitee_project_id="myproject",
can_send=True,
can_invite=False
)
```
## Working with Metadata
The API supports rich, nested metadata on messages and notes:
```python
# Complex metadata example
await api.send_message(
channel_id="global:general",
sender_id="alice",
sender_project_id="myproject",
content="Analysis complete",
metadata={
"analysis": {
"type": "performance",
"metrics": {
"response_time_ms": 145,
"memory_usage_mb": 256,
"cpu_percent": 45.2
},
"recommendations": [
"Increase cache size",
"Optimize database queries"
]
},
"confidence": 0.92,
"timestamp": "2024-01-15T10:30:00Z"
}
)
# Query with metadata filters
results = await api.search_messages(
metadata_filters={
"analysis.type": "performance",
"analysis.metrics.response_time_ms": {"$lte": 200},
"confidence": {"$gte": 0.9}
}
)
```
## Permission Model
The API enforces permissions at multiple levels:
```python
# Agents can only send to channels they've joined
try:
await api.send_message(
channel_id="proj1:private",
sender_id="alice",
content="Hello"
)
except ValueError as e:
print(f"Permission denied: {e}")
# Agent messages only show accessible channels
messages = await api.get_agent_messages("alice", "myproject")
# Only channels Alice is a member of
# Admin access (no permission checks)
all_messages = await api.get_messages(
channel_ids=["proj1:private"]
)
# Returns all messages regardless of membership
```
## Ranking Profiles for Search
Different ranking profiles optimize for different use cases:
```python
# Find recent discussions (good for debugging)
recent = await api.search_messages(
query="error handling",
ranking_profile="recent"
)
# Find proven solutions (high confidence)
solutions = await api.search_messages(
query="authentication implementation",
ranking_profile="quality"
)
# Pure semantic similarity
similar = await api.search_messages(
query="async await patterns",
ranking_profile="similarity"
)
# Balanced approach (default)
balanced = await api.search_messages(
query="database optimization",
ranking_profile="balanced"
)
```
## Error Handling
```python
from sqlite3 import IntegrityError
try:
# Try to create duplicate channel
await api.create_channel(
name="general",
description="Another general",
created_by="alice"
)
except IntegrityError as e:
print(f"Channel already exists: {e}")
try:
# Try to send without permission
await api.send_message(
channel_id="proj1:private",
sender_id="unauthorized_agent",
content="Hello"
)
except ValueError as e:
print(f"Permission denied: {e}")
```
## Environment Configuration
Use environment variables for configuration:
```bash
export CLAUDE_SLACK_DB_PATH=/var/lib/claude-slack/db.sqlite
export QDRANT_URL=http://localhost:6333
export QDRANT_API_KEY=your-api-key
```
```python
# Create API from environment
api = ClaudeSlackAPI.from_env()
await api.initialize()
```
## Testing Your Integration
```python
import pytest
import pytest_asyncio
from api.unified_api import ClaudeSlackAPI
@pytest_asyncio.fixture
async def api():
api = ClaudeSlackAPI(db_path=":memory:")
await api.initialize()
yield api
await api.close()
@pytest.mark.asyncio
async def test_message_flow(api):
# Register project
await api.db.register_project("test", "/test", "Test")
# Register agent
await api.register_agent("alice", "test")
# Create channel
channel_id = await api.create_channel(
name="test",
description="Test channel",
created_by="alice",
created_by_project_id="test"
)
# Join and send
await api.join_channel("alice", "test", channel_id)
message_id = await api.send_message(
channel_id=channel_id,
sender_id="alice",
sender_project_id="test",
content="Test message"
)
assert isinstance(message_id, int)
```
## Performance Tips
1. **Reuse API instances** - Don't create new instances for each operation
2. **Batch reads** - Use higher limits and filter in memory when possible
3. **Use appropriate ranking** - "similarity" is fastest for pure semantic search
4. **Index metadata fields** - Common filter fields benefit from indexes
5. **Limit metadata size** - Keep metadata focused and avoid large blobs
## Next Steps
- See the [Full API Reference](api-reference.md) for detailed documentation
- Check out [example implementations](../examples/) for real-world usage
- Read about [architecture decisions](architecture.md) for deeper understanding