Create Public API Endpoints for Streaming Team Responses and Retrieving Thread Data via API Key

README.md

@@ -42,7 +42,7 @@
 - [Contribution](#contribution)
 - [Release Notes](#release-notes)
 - [License](#license)
-  
+
 
 > [!WARNING]
 > This project is currently under heavy development. Please be aware that significant changes may occur.
@@ -66,6 +66,7 @@ and many many more!
 -  **Retrieval Augmented Generation**: Enable your agents to reason with your internal knowledge base.
 - **Human-In-The-Loop**: Enable human approval before tool calling.
 - **Open Source Models**: Use open-source LLM models such as llama, Gemma and Phi.
+- **Integrate Tribe with external application**: Use Tribe’s public API endpoints to interact with your teams.
 - **Easy Deployment**: Deploy Tribe effortlessly using Docker.
 - **Multi-Tenancy**: Manage and support multiple users and teams.
 
@@ -106,7 +107,7 @@ Copy the content and use that as password / secret key. And run that again to ge
 
 #### Sequential workflows
 
-In a sequential workflow, your agents are arranged in an orderly sequence and execute tasks one after another. Each task can be dependent on the previous task. This is useful if you want to tasks to be completed one after another in a deterministic sequence. 
+In a sequential workflow, your agents are arranged in an orderly sequence and execute tasks one after another. Each task can be dependent on the previous task. This is useful if you want to tasks to be completed one after another in a deterministic sequence.
 
 Use this if:
 - Your project has clear, step-by-step tasks.

backend/app/alembic/versions/25de3619cb35_add_apikeys_table.py

@@ -0,0 +1,38 @@
+"""add apikeys table
+
+Revision ID: 25de3619cb35
+Revises: 20f584dc80d2
+Create Date: 2024-08-31 07:19:42.927401
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+
+
+# revision identifiers, used by Alembic.
+revision = '25de3619cb35'
+down_revision = '20f584dc80d2'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('apikey',
+    sa.Column('description', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('hashed_key', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+    sa.Column('short_key', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+    sa.Column('team_id', sa.Integer(), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), nullable=True),
+    sa.ForeignKeyConstraint(['team_id'], ['team.id'], ),
+    sa.PrimaryKeyConstraint('id')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('apikey')
+    # ### end Alembic commands ###

backend/app/api/deps.py

@@ -3,15 +3,15 @@
 
 import jwt
 from fastapi import Depends, HTTPException, status
-from fastapi.security import OAuth2PasswordBearer
+from fastapi.security import APIKeyHeader, OAuth2PasswordBearer
 from jwt.exceptions import InvalidTokenError
 from pydantic import ValidationError
 from sqlmodel import Session
 
 from app.core import security
 from app.core.config import settings
 from app.core.db import engine
-from app.models import TokenPayload, User
+from app.models import Team, TokenPayload, User
 
 reusable_oauth2 = OAuth2PasswordBearer(
     tokenUrl=f"{settings.API_V1_STR}/login/access-token"
@@ -55,3 +55,28 @@ def get_current_active_superuser(current_user: CurrentUser) -> User:
             status_code=400, detail="The user doesn't have enough privileges"
         )
     return current_user
+
+
+header_scheme = APIKeyHeader(name="x-api-key")
+
+
+def get_current_team_from_key(
+    session: SessionDep,
+    team_id: int,
+    key: str = Depends(header_scheme),
+) -> Team:
+    """Return team if apikey belongs to it"""
+    team = session.get(Team, team_id)
+    if not team:
+        raise HTTPException(status_code=404, detail="Team not found")
+    verified = False
+    for apikey in team.apikeys:
+        if security.verify_password(key, apikey.hashed_key):
+            verified = True
+            break
+    if not verified:
+        raise HTTPException(status_code=401, detail="Invalid API key")
+    return team
+
+
+CurrentTeam = Annotated[Team, Depends(get_current_team_from_key)]

backend/app/api/main.py

@@ -1,6 +1,16 @@
 from fastapi import APIRouter
 
-from app.api.routes import login, members, skills, teams, threads, uploads, users, utils
+from app.api.routes import (
+    apikeys,
+    login,
+    members,
+    skills,
+    teams,
+    threads,
+    uploads,
+    users,
+    utils,
+)
 
 api_router = APIRouter()
 api_router.include_router(login.router, tags=["login"])
@@ -15,3 +25,6 @@
     threads.router, prefix="/teams/{team_id}/threads", tags=["threads"]
 )
 api_router.include_router(uploads.router, prefix="/uploads", tags=["uploads"])
+api_router.include_router(
+    apikeys.router, prefix="/teams/{team_id}/api-keys", tags=["api-keys"]
+)

backend/app/api/routes/apikeys.py

@@ -0,0 +1,121 @@
+from typing import Any
+
+from fastapi import APIRouter, HTTPException
+from sqlmodel import func, select
+
+from app.api.deps import CurrentUser, SessionDep
+from app.core.security import generate_apikey, generate_short_apikey, get_password_hash
+from app.models import ApiKey, ApiKeyCreate, ApiKeyOut, ApiKeysOutPublic, Message, Team
+
+router = APIRouter()
+
+
+@router.get("/", response_model=ApiKeysOutPublic)
+def read_api_keys(
+    session: SessionDep,
+    current_user: CurrentUser,
+    team_id: int,
+    skip: int = 0,
+    limit: int = 100,
+) -> Any:
+    """Read api keys"""
+    if current_user.is_superuser:
+        count_statement = select(func.count()).select_from(ApiKey)
+        count = session.exec(count_statement).one()
+        statement = (
+            select(ApiKey).where(ApiKey.team_id == team_id).offset(skip).limit(limit)
+        )
+        apikeys = session.exec(statement).all()
+    else:
+        count_statement = (
+            select(func.count())
+            .select_from(ApiKey)
+            .join(Team)
+            .where(Team.owner_id == current_user.id, ApiKey.team_id == team_id)
+        )
+        count = session.exec(count_statement).one()
+        statement = (
+            select(ApiKey)
+            .join(Team)
+            .where(Team.owner_id == current_user.id, ApiKey.team_id == team_id)
+            .offset(skip)
+            .limit(limit)
+        )
+        apikeys = session.exec(statement).all()
+
+    return ApiKeysOutPublic(data=apikeys, count=count)
+
+
+@router.post("/", response_model=ApiKeyOut)
+def create_api_key(
+    session: SessionDep,
+    current_user: CurrentUser,
+    team_id: int,
+    apikey_in: ApiKeyCreate,
+) -> Any:
+    """Create API key for a team."""
+
+    # Check if the current user is not a superuser
+    if not current_user.is_superuser:
+        # Validate the provided team_id and check ownership
+        team = session.get(Team, team_id)
+        if not team:
+            raise HTTPException(status_code=404, detail="Team not found.")
+        if team.owner_id != current_user.id:
+            raise HTTPException(status_code=403, detail="Not enough permissions.")
+
+    # Generate API key and hash it
+    key = generate_apikey()
+    hashed_key = get_password_hash(key)
+    short_key = generate_short_apikey(key)
+
+    # Create the API key object
+    description = apikey_in.description
+    apikey = ApiKey(
+        team_id=team_id,
+        hashed_key=hashed_key,
+        short_key=short_key,
+        description=None if not description or not description.strip() else description,
+    )
+
+    # Save the new API key to the database
+    session.add(apikey)
+    session.commit()
+    session.refresh(apikey)
+
+    return ApiKeyOut(
+        id=apikey.id,
+        description=apikey.description,
+        key=key,
+        created_at=apikey.created_at,
+    )
+
+
+@router.delete("/{id}")
+def delete_api_key(
+    session: SessionDep, current_user: CurrentUser, team_id: int, id: int
+) -> Any:
+    """Delete API key for a team."""
+    if current_user.is_superuser:
+        statement = (
+            select(ApiKey).join(Team).where(ApiKey.id == id, ApiKey.team_id == team_id)
+        )
+        apikey = session.exec(statement).first()
+    else:
+        statement = (
+            select(ApiKey)
+            .join(Team)
+            .where(
+                ApiKey.id == id,
+                ApiKey.team_id == team_id,
+                Team.owner_id == current_user.id,
+            )
+        )
+        apikey = session.exec(statement).first()
+
+    if not apikey:
+        raise HTTPException(status_code=404, detail="Api key not found")
+
+    session.delete(apikey)
+    session.commit()
+    return Message(message="Api key deleted successfully")

backend/app/api/routes/teams.py

@@ -1,16 +1,23 @@
+from datetime import datetime
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException
 from fastapi.responses import StreamingResponse
+from fastapi.security import APIKeyHeader
 from sqlmodel import col, func, select
 
-from app.api.deps import CurrentUser, SessionDep
+from app.api.deps import (
+    CurrentTeam,
+    CurrentUser,
+    SessionDep,
+)
 from app.core.graph.build import generator
 from app.models import (
     Member,
     Message,
     Team,
     TeamChat,
+    TeamChatPublic,
     TeamCreate,
     TeamOut,
     TeamsOut,
@@ -20,6 +27,8 @@
 
 router = APIRouter()
 
+header_scheme = APIKeyHeader(name="x-api-key")
+
 
 async def validate_name_on_create(session: SessionDep, team_in: TeamCreate) -> None:
     """Validate that team name is unique"""
@@ -206,3 +215,68 @@ async def stream(
         generator(team, members, team_chat.messages, thread_id, team_chat.interrupt),
         media_type="text/event-stream",
     )
+
+
+@router.post("/{team_id}/stream-public/{thread_id}")
+async def public_stream(
+    session: SessionDep,
+    team_id: int,
+    team_chat: TeamChatPublic,
+    thread_id: str,
+    team: CurrentTeam,
+) -> StreamingResponse:
+    """
+    Stream a response from a team using a given message or an interrupt decision. Requires an API key for authentication.
+
+    This endpoint allows streaming responses from a team based on a provided message or interrupt details. The request must include an API key for authorization.
+
+    Parameters:
+    - `team_id` (int): The ID of the team to which the message is being sent. Must be a valid team ID.
+    - `thread_id` (str): The ID of the thread where the message will be posted. If the thread ID does not exist, a new thread will be created.
+
+    Request Body (JSON):
+    - The request body should be a JSON object containing either the `message` or `interrupt` field:
+        - `message` (object, optional): The message to be sent to the team.
+            - `type` (str): Must be `"human"`.
+            - `content` (str): The content of the message to be sent.
+        - `interrupt` (object, optional): Approve/reject tool or reply to an ask-human tool.
+            - `decision` (str): Can be `'approved'`, `'rejected'`, or `'replied'`.
+            - `tool_message` (str or null, optional): If `decision` is `'rejected'` or `'replied'`, provide a message explaining the reason for rejection or the reply.
+
+    Authorization:
+    - API key must be provided in the request header as `x-api-key`.
+
+    Responses:
+    - `200 OK`: Returns a streaming response in `text/event-stream` format containing the team's response.
+    """
+    # Check if thread belongs to the team
+    thread = session.get(Thread, thread_id)
+    message_content = team_chat.message.content if team_chat.message else ""
+    if not thread:
+        # create new thread
+        thread = Thread(
+            id=thread_id,
+            query=message_content,
+            updated_at=datetime.now(),
+            team_id=team_id,
+        )
+        session.add(thread)
+        session.commit()
+        session.refresh(thread)
+    else:
+        if thread.team_id != team_id:
+            raise HTTPException(
+                status_code=400, detail="Thread does not belong to the team"
+            )
+
+    # Populate the skills and accessible uploads for each member
+    members = team.members
+    for member in members:
+        member.skills = member.skills
+        member.uploads = member.uploads
+
+    messages = [team_chat.message] if team_chat.message else []
+    return StreamingResponse(
+        generator(team, members, messages, thread_id, team_chat.interrupt),
+        media_type="text/event-stream",
+    )