bedRockHard

A small proxy to pass Claude API-style requests to Amazon Bedrock, specifically for claude-2.

Some notes:

I don't know if you need an AWS business account for that or an individual one will work.
As of now, boto is in a weird state and the one you get from pip might be the old one which doesn't have bedrock-runtime yet. So if your aioboto install doesn't work, your best bet is to wait for a few days for them to update the SDKs properly, or patch the things manually.
You have to request access to Claude in https://<region>.console.aws.amazon.com/bedrock/home?region=<region>#/modelaccess ("Model Access" on the left sidebar on the Amazon Bedrock page). There's either a bug or something - for "Anthropic" section there's the "Request" button that asks you to fill all forms for Anthropic to give you access, and the "Claude Instant" model is greyed out (presumably until you fill that form), but! "Claude" (which is 1.3 and 2) can be enabled without filling that form anyway. If it's a bug - it'll probably get fixed real soon, if not - good.
You have to give your IAM account (the one you'll do the requests with) bedrock permissions - at least InvokeModel and InvokeModelWithResponseStream

For more info see Amazon Bedrock User Guide

Requirements: pip install aiohttp aiobotocore (but see note above about boto)

Proxy (after you start it, put http://localhost:8123/v1 in the ST reverse proxy, of course with Claude selected):

⎗
✓
import json

from contextlib import AsyncExitStack

from aiohttp import web
from aiobotocore.session import AioSession

exit_stack = AsyncExitStack()
session = AioSession()
client = None

aws_id = "" # AKIA.... shit
aws_key = "" # long secret key
aws_region = "us-east-1" # default region 

# No clue if it's actually safe to use the same client for async reqs or not,
# but should work for basic single-user usage.
async def create_client():
    if client is None:
        return await exit_stack.enter_async_context(session.create_client(
            aws_access_key_id=aws_id,
            aws_secret_access_key=aws_key,
            service_name="bedrock-runtime",  # yes, specifically this, NOT "bedrock"!
            region_name=aws_region
        ))
    else:
        return client


async def handle(request):
    client = await create_client()
    data = await request.json()
    stream = data.get("stream", False)
    # Skip stream and model params
    claude_body = json.dumps({k: v for k, v in data.items() if k not in ["stream", "model"]})

    modelId = "anthropic.claude-v2"
    accept = "application/json"
    contentType = "application/json"

    if not stream:
        # Basic request, just forward the resp
        resp = await client.invoke_model(body=claude_body, modelId=modelId, accept=accept, contentType=contentType)
        body = await resp["body"].read()
        return web.json_response(json.loads(body))

    else:
        response = web.StreamResponse()
        response.headers["Content-Type"] = "text/event-stream"
        response.headers["Cache-Control"] = "no-cache"
        response.headers["Connection"] = "keep-alive"
        await response.prepare(request)

        invoke_resp = await client.invoke_model_with_response_stream(modelId=modelId, body=claude_body)
        stream = invoke_resp.get("body")
        if stream:
            async for event in stream:
                chunk = event.get("chunk")
                if chunk:
                    chunk_data = json.loads(chunk.get("bytes").decode())
                    message = f"event: completion\ndata: {json.dumps(chunk_data)}\n\n"
                    await response.write(message.encode("utf-8"))

        await response.write_eof()
        return response

app = web.Application()
app.add_routes([web.post("/v1/complete", handle)])

web.run_app(app, port=8123)

CncAnon, 2023